You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@toree.apache.org by lb...@apache.org on 2016/01/22 23:07:28 UTC
[20/51] [abbrv] incubator-toree git commit: Moved scala files to new
locations based on new package
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel-api/src/test/scala/org/apache/toree/utils/ScheduledTaskManagerSpec.scala
----------------------------------------------------------------------
diff --git a/kernel-api/src/test/scala/org/apache/toree/utils/ScheduledTaskManagerSpec.scala b/kernel-api/src/test/scala/org/apache/toree/utils/ScheduledTaskManagerSpec.scala
new file mode 100644
index 0000000..8db4535
--- /dev/null
+++ b/kernel-api/src/test/scala/org/apache/toree/utils/ScheduledTaskManagerSpec.scala
@@ -0,0 +1,129 @@
+/*
+ * Copyright 2014 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.ibm.spark.utils
+
+import java.util.Calendar
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.{TimeUnit, CountDownLatch}
+
+import org.scalatest.concurrent.Eventually
+import org.scalatest.time.{Milliseconds, Span}
+import org.scalatest.{BeforeAndAfter, Matchers, FunSpec}
+
+class ScheduledTaskManagerSpec extends FunSpec with Matchers with BeforeAndAfter
+ with Eventually
+{
+ private val TestTimeInterval = 30
+ private val MaximumChecks = 3
+ private val TimeoutScale = 3
+ private var scheduledTaskManager: ScheduledTaskManager = _
+ private var scheduleVerifier: ScheduleVerifier = _
+
+ implicit override val patienceConfig = PatienceConfig(
+ timeout = scaled(Span(
+ TestTimeInterval * MaximumChecks * TimeoutScale, Milliseconds)),
+ interval = scaled(Span(TestTimeInterval / 2, Milliseconds))
+ )
+
+ private class ScheduleVerifier {
+ @volatile private var checkinTimes: List[Long] = Nil
+ private val taskRun = new AtomicBoolean(false)
+
+ def task() = {
+ if (checkinTimes.length < MaximumChecks)
+ checkinTimes = checkinTimes :+ Calendar.getInstance().getTimeInMillis
+ taskRun.set(true)
+ }
+
+ def shouldNotRunAnymore(milliseconds: Long) = eventually {
+ val offset: Int = (milliseconds * 0.5).toInt
+
+ // Clean the state and wait to see if the task is executed again
+ taskRun.set(false)
+ Thread.sleep(milliseconds + offset)
+
+ taskRun.get() should be (false)
+ }
+
+ def shouldRunEvery(milliseconds: Long) = {
+ // 50% +/-
+ val offset: Int = (milliseconds * 0.5).toInt
+
+ eventually {
+ // Assert we have the desired number of checks
+ checkinTimes.length should be (MaximumChecks)
+
+ checkinTimes.take(checkinTimes.length - 1).zip(
+ checkinTimes.takeRight(checkinTimes.length - 1)
+ ).foreach({ times =>
+ val firstTime = times._1
+ val secondTime = times._2
+ (secondTime - firstTime) should (
+ be >= milliseconds - offset and
+ be <= milliseconds + offset)
+ })
+ }
+ }
+ }
+
+ before {
+ scheduledTaskManager = new ScheduledTaskManager
+ scheduleVerifier = new ScheduleVerifier
+ }
+
+ after {
+ scheduledTaskManager.stop()
+ }
+
+ describe("ScheduledTaskManager") {
+ describe("#addTask") {
+ // TODO: This is failing frequently due to some sort of timing problem
+ ignore("should add a new task to be executed periodically") {
+ scheduledTaskManager.addTask(timeInterval = TestTimeInterval,
+ task = scheduleVerifier.task())
+
+ scheduleVerifier.shouldRunEvery(TestTimeInterval)
+ }
+ }
+
+ describe("#removeTask") {
+ it("should stop and remove the task if it exists") {
+ val taskId = scheduledTaskManager.addTask(
+ timeInterval = TestTimeInterval,
+ task = scheduleVerifier.task()
+ )
+
+ scheduledTaskManager.removeTask(taskId)
+
+ scheduleVerifier.shouldNotRunAnymore(TestTimeInterval)
+ }
+
+ it("should return true if the task was removed") {
+ val taskId = scheduledTaskManager.addTask(
+ timeInterval = TestTimeInterval,
+ task = scheduleVerifier.task()
+ )
+
+ scheduledTaskManager.removeTask(taskId) should be (true)
+ }
+
+ it("should return false if the task does not exist") {
+ scheduledTaskManager.removeTask("") should be (false)
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel-api/src/test/scala/org/apache/toree/utils/TaskManagerSpec.scala
----------------------------------------------------------------------
diff --git a/kernel-api/src/test/scala/org/apache/toree/utils/TaskManagerSpec.scala b/kernel-api/src/test/scala/org/apache/toree/utils/TaskManagerSpec.scala
new file mode 100644
index 0000000..3456d98
--- /dev/null
+++ b/kernel-api/src/test/scala/org/apache/toree/utils/TaskManagerSpec.scala
@@ -0,0 +1,301 @@
+/*
+ * Copyright 2014 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.ibm.spark.utils
+
+import java.util.concurrent.{RejectedExecutionException, ExecutionException}
+
+import org.scalatest.concurrent.PatienceConfiguration.Timeout
+import org.scalatest.concurrent.{Timeouts, Eventually, ScalaFutures}
+import org.scalatest.mock.MockitoSugar
+import org.scalatest.time.{Milliseconds, Seconds, Span}
+import org.scalatest.{BeforeAndAfter, FunSpec, Matchers}
+import test.utils.UncaughtExceptionSuppression
+
+import scala.concurrent.Future
+import scala.runtime.BoxedUnit
+
+class TaskManagerSpec extends FunSpec with Matchers with MockitoSugar
+ with BeforeAndAfter with ScalaFutures with UncaughtExceptionSuppression
+ with Eventually with Timeouts
+{
+ implicit override val patienceConfig = PatienceConfig(
+ timeout = scaled(Span(200, Milliseconds)),
+ interval = scaled(Span(5, Milliseconds))
+ )
+ private val MaxTestTasks = 50000
+ private var taskManager: TaskManager = _
+
+ before {
+ taskManager = new TaskManager
+ }
+
+ after {
+ taskManager = null
+ }
+
+ describe("TaskManager") {
+ describe("#add") {
+ it("should throw an exception if not started") {
+ intercept[AssertionError] {
+ taskManager.add {}
+ }
+ }
+
+ it("should throw an exception if more tasks are added than max task size") {
+ val taskManager = new TaskManager(maximumWorkers = 1, maxTasks = 1)
+
+ taskManager.start()
+
+ // Should fail from having too many tasks added
+ intercept[RejectedExecutionException] {
+ for (i <- 1 to MaxTestTasks) taskManager.add {}
+ }
+ }
+
+ it("should return a Future[_] based on task provided") {
+ taskManager.start()
+
+ // Cannot check inner Future type due to type erasure
+ taskManager.add { } shouldBe an [Future[_]]
+
+ taskManager.stop()
+ }
+
+ it("should work for a task that returns nothing") {
+ taskManager.start()
+
+ val f = taskManager.add { }
+
+ whenReady(f) { result =>
+ result shouldBe a [BoxedUnit]
+ taskManager.stop()
+ }
+ }
+
+ it("should construct a Runnable that invokes a Promise on success") {
+ taskManager.start()
+
+ val returnValue = 3
+ val f = taskManager.add { returnValue }
+
+ whenReady(f) { result =>
+ result should be (returnValue)
+ taskManager.stop()
+ }
+ }
+
+ it("should construct a Runnable that invokes a Promise on failure") {
+ taskManager.start()
+
+ val error = new Throwable("ERROR")
+ val f = taskManager.add { throw error }
+
+ whenReady(f.failed) { result =>
+ result should be (error)
+ taskManager.stop()
+ }
+ }
+
+ it("should not block when adding more tasks than available threads") {
+ val taskManager = new TaskManager(maximumWorkers = 1)
+
+ taskManager.start()
+
+ failAfter(Span(100, Milliseconds)) {
+ taskManager.add { while (true) { Thread.sleep(1) } }
+ taskManager.add { while (true) { Thread.sleep(1) } }
+ }
+ }
+ }
+
+ describe("#size") {
+ it("should be zero when no tasks have been added") {
+ taskManager.size should be (0)
+ }
+
+ it("should reflect queued tasks and executing tasks") {
+ val taskManager = new TaskManager(maximumWorkers = 1)
+ taskManager.start()
+
+ // Fill up the task manager and then add another task to the queue
+ taskManager.add { while (true) { Thread.sleep(1000) } }
+ taskManager.add { while (true) { Thread.sleep(1000) } }
+
+ taskManager.size should be (2)
+ }
+
+ it("should be one if there is only one executing task and no queued ones") {
+ taskManager.start()
+
+ taskManager.add { while (true) { Thread.sleep(1000) } }
+
+ // Wait until task is being executed to check if the task is still in
+ // the queue
+ while (!taskManager.isExecutingTask) Thread.sleep(1)
+
+ taskManager.size should be (1)
+
+ taskManager.stop()
+ }
+ }
+
+ describe("#hasTaskInQueue") {
+ it("should be false when no task has been added") {
+ taskManager.hasTaskInQueue should be (false)
+ }
+
+ it("should be true where there are tasks remaining in the queue") {
+ val taskManager = new TaskManager(maximumWorkers = 1)
+ taskManager.start()
+
+ // Fill up the task manager and then add another task to the queue
+ taskManager.add { while (true) { Thread.sleep(1000) } }
+ taskManager.add { while (true) { Thread.sleep(1000) } }
+
+ taskManager.hasTaskInQueue should be (true)
+ }
+
+ it("should be false when the only task is currently being executed") {
+ taskManager.start()
+
+ taskManager.add { while (true) { Thread.sleep(1000) } }
+
+ // Wait until task is being executed to check if the task is still in
+ // the queue
+ while (!taskManager.isExecutingTask) Thread.sleep(1)
+
+ taskManager.hasTaskInQueue should be (false)
+
+ taskManager.stop()
+ }
+ }
+
+ describe("#isExecutingTask") {
+ it("should be true when a task is being executed") {
+ taskManager.start()
+ taskManager.add { while (true) { Thread.sleep(1000) } }
+
+ eventually {
+ taskManager.isExecutingTask should be (true)
+ }
+
+ taskManager.stop()
+ }
+
+ it("should be false when no tasks have been added") {
+ taskManager.isExecutingTask should be (false)
+ }
+
+ // TODO: Timing issue on Travis CI needs to be resolved
+ ignore("should be false when all tasks have finished") {
+ taskManager.start()
+ val f = taskManager.add { } // Really fast execution
+
+ // Wait for up to 1 second for the task to finish
+ whenReady(f, Timeout(Span(1, Seconds))) { result =>
+ taskManager.isExecutingTask should be (false)
+ taskManager.stop()
+ }
+ }
+ }
+
+ describe("#await") {
+ it("should block until all tasks are completed") {
+ val taskManager = new TaskManager(
+ maximumWorkers = 1,
+ maxTasks = MaxTestTasks
+ )
+
+ taskManager.start()
+
+ // TODO: Need better way to ensure tasks are still running while
+ // awaiting their return
+ for (x <- 1 to MaxTestTasks) taskManager.add { Thread.sleep(1) }
+
+ assume(taskManager.hasTaskInQueue)
+ taskManager.await()
+
+ taskManager.hasTaskInQueue should be (false)
+ taskManager.isExecutingTask should be (false)
+
+ taskManager.stop()
+ }
+ }
+
+ describe("#start") {
+ it("should create an internal thread pool executor") {
+ taskManager.start()
+
+ taskManager.executor should not be (None)
+
+ taskManager.stop()
+ }
+ }
+
+ describe("#restart") {
+ it("should stop & erase the old internal thread and create a new one") {
+ taskManager.start()
+
+ val oldExecutor = taskManager.executor
+
+ taskManager.restart()
+
+ taskManager.executor should not be (oldExecutor)
+
+ taskManager.stop()
+ }
+ }
+
+ describe("#stop") {
+ it("should attempt to interrupt the currently-running task") {
+ taskManager.start()
+ val f = taskManager.add { while (true) { Thread.sleep(1000) } }
+
+ // Wait for the task to start
+ while (!taskManager.isExecutingTask) Thread.sleep(1)
+
+ // Cancel the task
+ taskManager.stop()
+
+ // Future should return an InterruptedException
+ whenReady(f.failed) { result =>
+ result shouldBe an [ExecutionException]
+ result.getCause shouldBe an [InterruptedException]
+ }
+ }
+
+ // TODO: Refactoring task manager to be parallelizable broke this ability
+ // so this will need to be reimplemented or abandoned
+ ignore("should kill the thread if interrupts failed and kill enabled") {
+ taskManager.start()
+ val f = taskManager.add { var x = 0; while (true) { x += 1 } }
+
+ // Wait for the task to start
+ while (!taskManager.isExecutingTask) Thread.sleep(1)
+
+ // Kill the task
+ taskManager.stop()
+
+ // Future should return ThreadDeath when killed
+ whenReady(f.failed) { result =>
+ result shouldBe an [ExecutionException]
+ result.getCause shouldBe a [ThreadDeath]
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel/src/main/scala/com/ibm/spark/SparkKernel.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/SparkKernel.scala b/kernel/src/main/scala/com/ibm/spark/SparkKernel.scala
deleted file mode 100644
index f532de9..0000000
--- a/kernel/src/main/scala/com/ibm/spark/SparkKernel.scala
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * Copyright 2014 IBM Corp.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.ibm.spark
-
-import com.ibm.spark.boot.layer._
-import com.ibm.spark.boot.{CommandLineOptions, KernelBootstrap}
-import com.ibm.spark.kernel.BuildInfo
-
-object SparkKernel extends App {
- private val options = new CommandLineOptions(args)
-
- if (options.help) {
- options.printHelpOn(System.out)
- } else if (options.version) {
- println(s"Kernel Version: ${BuildInfo.version}")
- println(s"Build Date: ${BuildInfo.buildDate}")
- println(s"Scala Version: ${BuildInfo.scalaVersion}")
- println(s"Apache Spark Version: ${BuildInfo.sparkVersion}")
- } else {
- (new KernelBootstrap(options.toConfig)
- with StandardBareInitialization
- with StandardComponentInitialization
- with StandardHandlerInitialization
- with StandardHookInitialization)
- .initialize()
- .waitForTermination()
- .shutdown()
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel/src/main/scala/com/ibm/spark/boot/CommandLineOptions.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/boot/CommandLineOptions.scala b/kernel/src/main/scala/com/ibm/spark/boot/CommandLineOptions.scala
deleted file mode 100644
index a5acbc2..0000000
--- a/kernel/src/main/scala/com/ibm/spark/boot/CommandLineOptions.scala
+++ /dev/null
@@ -1,199 +0,0 @@
-/*
- * Copyright 2015 IBM Corp.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.ibm.spark.boot
-
-import java.io.{File, OutputStream}
-
-import com.ibm.spark.utils.KeyValuePairUtils
-import com.typesafe.config.{Config, ConfigFactory}
-import joptsimple.util.KeyValuePair
-import joptsimple.{OptionParser, OptionSpec}
-
-import scala.collection.JavaConverters._
-
-class CommandLineOptions(args: Seq[String]) {
- private val parser = new OptionParser()
- parser.allowsUnrecognizedOptions()
-
- /*
- * Options supported by our kernel.
- */
- private val _help =
- parser.acceptsAll(Seq("help", "h").asJava, "display help information").forHelp()
-
- private val _version =
- parser.acceptsAll(Seq("version", "v").asJava, "display version information")
-
- private val _profile =
- parser.accepts("profile", "path to IPython JSON connection file")
- .withRequiredArg().ofType(classOf[File])
-
- private val _ip =
- parser.accepts("ip", "ip used to bind sockets")
- .withRequiredArg().ofType(classOf[String])
-
- private val _stdin_port = parser.accepts(
- "stdin-port", "port of the stdin socket"
- ).withRequiredArg().ofType(classOf[Int])
-
- private val _shell_port = parser.accepts(
- "shell-port", "port of the shell socket"
- ).withRequiredArg().ofType(classOf[Int])
-
- private val _iopub_port = parser.accepts(
- "iopub-port", "port of the iopub socket"
- ).withRequiredArg().ofType(classOf[Int])
-
- private val _control_port = parser.accepts(
- "control-port", "port of the control socket"
- ).withRequiredArg().ofType(classOf[Int])
-
- private val _heartbeat_port = parser.accepts(
- "heartbeat-port", "port of the heartbeat socket"
- ).withRequiredArg().ofType(classOf[Int])
-
- private val _spark_configuration = parser.acceptsAll(
- Seq("spark-configuration", "S").asJava,
- "configuration setting for Apache Spark"
- ).withRequiredArg().ofType(classOf[KeyValuePair])
-
- private val _magic_url =
- parser.accepts("magic-url", "path to a magic jar")
- .withRequiredArg().ofType(classOf[String])
-
- private val _max_interpreter_threads = parser.accepts(
- "max-interpreter-threads",
- "total number of worker threads to use to execute code"
- ).withRequiredArg().ofType(classOf[Int])
-
- private val _jar_dir = parser.accepts(
- "jar-dir",
- "directory where user added jars are stored (MUST EXIST)"
- ).withRequiredArg().ofType(classOf[String])
-
- private val _default_interpreter =
- parser.accepts("default-interpreter", "default interpreter for the kernel")
- .withRequiredArg().ofType(classOf[String])
-
- private val _nosparkcontext =
- parser.accepts("nosparkcontext", "kernel should not create a spark context")
-
- private val _interpreter_plugin = parser.accepts(
- "interpreter-plugin"
- ).withRequiredArg().ofType(classOf[String])
-
- private val options = parser.parse(args.map(_.trim): _*)
-
- /*
- * Helpers to determine if an option is provided and the value with which it
- * was provided.
- */
-
- private def has[T](spec: OptionSpec[T]): Boolean =
- options.has(spec)
-
- private def get[T](spec: OptionSpec[T]): Option[T] =
- Some(options.valueOf(spec)).filter(_ != null)
-
- private def getAll[T](spec: OptionSpec[T]): Option[List[T]] =
- Some(options.valuesOf(spec).asScala.toList).filter(_ != null)
-
- /*
- * Expose options in terms of their existence/value.
- */
-
- val help: Boolean = has(_help)
-
- val version: Boolean = has(_version)
-
- /*
- * Config object has 3 levels and fallback in this order
- * 1. Comandline Args
- * 2. --profile file
- * 3. Defaults
- */
- def toConfig: Config = {
- val profileConfig: Config = get(_profile) match {
- case Some(x) =>
- ConfigFactory.parseFile(x)
- case None =>
- ConfigFactory.empty()
- }
-
- val commandLineConfig: Config = ConfigFactory.parseMap(Map(
- "stdin_port" -> get(_stdin_port),
- "shell_port" -> get(_shell_port),
- "iopub_port" -> get(_iopub_port),
- "control_port" -> get(_control_port),
- "hb_port" -> get(_heartbeat_port),
- "ip" -> get(_ip),
- "interpreter_args" -> interpreterArgs,
- "magic_urls" -> getAll(_magic_url).map(_.asJava)
- .flatMap(list => if (list.isEmpty) None else Some(list)),
- "spark_configuration" -> getAll(_spark_configuration)
- .map(list => KeyValuePairUtils.keyValuePairSeqToString(list))
- .flatMap(str => if (str.nonEmpty) Some(str) else None),
- "max_interpreter_threads" -> get(_max_interpreter_threads),
- "jar_dir" -> get(_jar_dir),
- "default_interpreter" -> get(_default_interpreter),
- "nosparkcontext" -> (if (has(_nosparkcontext)) Some(true) else Some(false)),
- "interpreter_plugins" -> interpreterPlugins
- ).flatMap(removeEmptyOptions).asInstanceOf[Map[String, AnyRef]].asJava)
-
- commandLineConfig.withFallback(profileConfig).withFallback(ConfigFactory.load)
- }
-
- private val removeEmptyOptions: ((String, Option[Any])) => Iterable[(String, Any)] = {
- pair => if (pair._2.isDefined) Some((pair._1, pair._2.get)) else None
- }
-
- /**
- *
- * @return
- */
- private def interpreterArgs: Option[java.util.List[String]] = {
- args.dropWhile(_ != "--").drop(1).toList match {
- case Nil => None
- case list: List[String] => Some(list.asJava)
- }
- }
-
- private def interpreterPlugins: Option[java.util.List[String]] = {
- //val defaults = getAll(_default_interpreter_plugin).getOrElse(List())
- //val defaults = List[String](
- // "PySpark:com.ibm.spark.kernel.interpreter.pyspark.PySparkInterpreter",
- // "SparkR:com.ibm.spark.kernel.interpreter.sparkr.SparkRInterpreter",
- // "SQL:com.ibm.spark.kernel.interpreter.sql.SqlInterpreter"
- //)
-
- val userDefined = getAll(_interpreter_plugin) match {
- case Some(l) => l
- case _ => List[String]()
- }
-
- //val p = defaults ++ userDefined
- Some(userDefined.asJava)
- }
-
- /**
- * Prints the help message to the output stream provided.
- * @param out The output stream to direct the help message
- */
- def printHelpOn(out: OutputStream) =
- parser.printHelpOn(out)
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel/src/main/scala/com/ibm/spark/boot/KernelBootstrap.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/boot/KernelBootstrap.scala b/kernel/src/main/scala/com/ibm/spark/boot/KernelBootstrap.scala
deleted file mode 100644
index 1e7927c..0000000
--- a/kernel/src/main/scala/com/ibm/spark/boot/KernelBootstrap.scala
+++ /dev/null
@@ -1,172 +0,0 @@
-/*
- * Copyright 2014 IBM Corp.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.ibm.spark.boot
-
-import akka.actor.{ActorRef, ActorSystem}
-import com.ibm.spark.boot.layer._
-import com.ibm.spark.interpreter.Interpreter
-import com.ibm.spark.kernel.api.Kernel
-import com.ibm.spark.kernel.protocol.v5.KernelStatusType._
-import com.ibm.spark.kernel.protocol.v5._
-import com.ibm.spark.kernel.protocol.v5.kernel.ActorLoader
-import com.ibm.spark.security.KernelSecurityManager
-import com.ibm.spark.utils.LogLike
-import com.typesafe.config.Config
-import org.apache.spark.SparkContext
-import org.zeromq.ZMQ
-
-import scala.util.Try
-
-class KernelBootstrap(config: Config) extends LogLike {
- this: BareInitialization with ComponentInitialization
- with HandlerInitialization with HookInitialization =>
-
- private val DefaultAppName = SparkKernelInfo.banner
- private val DefaultActorSystemName = "spark-kernel-actor-system"
-
- private var actorSystem: ActorSystem = _
- private var actorLoader: ActorLoader = _
- private var kernelMessageRelayActor: ActorRef = _
- private var statusDispatch: ActorRef = _
- private var kernel: Kernel = _
-
- private var sparkContext: SparkContext = _
- private var interpreters: Seq[Interpreter] = Nil
-
- /**
- * Initializes all kernel systems.
- */
- def initialize() = {
- // TODO: Investigate potential to initialize System out/err/in to capture
- // Console DynamicVariable initialization (since takes System fields)
- // and redirect it to a workable location (like an actor) with the
- // thread's current information attached
- //
- // E.G. System.setOut(customPrintStream) ... all new threads will have
- // customPrintStream as their initial Console.out value
- //
-
- displayVersionInfo()
-
- // Do this first to support shutting down quickly before entire system
- // is ready
- initializeShutdownHook()
-
- // Initialize the bare minimum to report a starting message
- val (actorSystem, actorLoader, kernelMessageRelayActor, statusDispatch) =
- initializeBare(
- config = config,
- actorSystemName = DefaultActorSystemName
- )
-
- this.actorSystem = actorSystem
- this.actorLoader = actorLoader
- this.kernelMessageRelayActor = kernelMessageRelayActor
- this.statusDispatch = statusDispatch
-
- // Indicate that the kernel is now starting
- publishStatus(KernelStatusType.Starting)
-
- // Initialize components needed elsewhere
- val (commStorage, commRegistrar, commManager, interpreter,
- kernel, dependencyDownloader,
- magicLoader, responseMap) =
- initializeComponents(
- config = config,
- appName = DefaultAppName,
- actorLoader = actorLoader
- )
- //this.sparkContext = sparkContext
- this.interpreters ++= Seq(interpreter)
-
- this.kernel = kernel
-
- // Initialize our handlers that take care of processing messages
- initializeHandlers(
- actorSystem = actorSystem,
- actorLoader = actorLoader,
- kernel = kernel,
- interpreter = interpreter,
- commRegistrar = commRegistrar,
- commStorage = commStorage,
- magicLoader = magicLoader,
- responseMap = responseMap
- )
-
- // Initialize our non-shutdown hooks that handle various JVM events
- initializeHooks(
- interpreter = interpreter
- )
-
- logger.debug("Initializing security manager")
- System.setSecurityManager(new KernelSecurityManager)
-
- logger.info("Marking relay as ready for receiving messages")
- kernelMessageRelayActor ! true
-
- this
- }
-
- /**
- * Shuts down all kernel systems.
- */
- def shutdown() = {
- logger.info("Shutting down Spark Context")
- Try(kernel.sparkContext.stop()).failed.foreach(
- logger.error("Failed to shutdown Spark Context", _: Throwable)
- )
-
- logger.info("Shutting down interpreters")
- Try(interpreters.foreach(_.stop())).failed.foreach(
- logger.error("Failed to shutdown interpreters", _: Throwable)
- )
-
- logger.info("Shutting down actor system")
- Try(actorSystem.shutdown()).failed.foreach(
- logger.error("Failed to shutdown actor system", _: Throwable)
- )
-
- this
- }
-
- /**
- * Waits for the main actor system to terminate.
- */
- def waitForTermination() = {
- logger.debug("Waiting for actor system to terminate")
- actorSystem.awaitTermination()
-
- this
- }
-
- private def publishStatus(
- status: KernelStatusType,
- parentHeader: Option[ParentHeader] = None
- ): Unit = {
- parentHeader match {
- case Some(header) => statusDispatch ! ((status, header))
- case None => statusDispatch ! status
- }
- }
-
- @inline private def displayVersionInfo() = {
- logger.info("Kernel version: " + SparkKernelInfo.implementationVersion)
- logger.info("Scala version: " + SparkKernelInfo.languageVersion)
- logger.info("ZeroMQ (JeroMQ) version: " + ZMQ.getVersionString)
- }
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel/src/main/scala/com/ibm/spark/boot/layer/BareInitialization.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/boot/layer/BareInitialization.scala b/kernel/src/main/scala/com/ibm/spark/boot/layer/BareInitialization.scala
deleted file mode 100644
index d2d6ab9..0000000
--- a/kernel/src/main/scala/com/ibm/spark/boot/layer/BareInitialization.scala
+++ /dev/null
@@ -1,181 +0,0 @@
-/*
- * Copyright 2014 IBM Corp.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.ibm.spark.boot.layer
-
-import akka.actor.{ActorRef, Props, ActorSystem}
-import com.ibm.spark.kernel.protocol.v5.dispatch.StatusDispatch
-import com.ibm.spark.kernel.protocol.v5.handler.{GenericSocketMessageHandler, ShutdownHandler}
-import com.ibm.spark.kernel.protocol.v5.kernel.{SimpleActorLoader, ActorLoader}
-import com.ibm.spark.communication.security.{SecurityActorType, SignatureManagerActor}
-import com.ibm.spark.kernel.protocol.v5.kernel.socket._
-import com.ibm.spark.kernel.protocol.v5._
-import com.ibm.spark.kernel.protocol.v5.content.{CommClose, CommMsg, CommOpen}
-import com.ibm.spark.kernel.protocol.v5.relay.KernelMessageRelay
-import com.ibm.spark.utils.LogLike
-import com.typesafe.config.Config
-import play.api.libs.json.Json
-
-/**
- * Represents the raw initialization needed to send a "starting" message.
- */
-trait BareInitialization {
- /**
- * Initializes and registers all objects needed to get the kernel to send a
- * "starting" message.
- *
- * @param config The config used for initialization
- * @param actorSystemName The name to use for the actor system
- */
- def initializeBare(config: Config, actorSystemName: String):
- (ActorSystem, ActorLoader, ActorRef, ActorRef)
-}
-
-/**
- * Represents the standard implementation of BareInitialization.
- */
-trait StandardBareInitialization extends BareInitialization { this: LogLike =>
- /**
- * Initializes and registers all objects needed to get the kernel to send a
- * "starting" message.
- *
- * @param config The config used for initialization
- * @param actorSystemName The name to use for the actor system
- */
- def initializeBare(config: Config, actorSystemName: String) = {
- val actorSystem = createActorSystem(actorSystemName)
- val actorLoader = createActorLoader(actorSystem)
- val (kernelMessageRelayActor, _, statusDispatch, _, _) =
- initializeCoreActors(config, actorSystem, actorLoader)
- createSockets(config, actorSystem, actorLoader)
-
- (actorSystem, actorLoader, kernelMessageRelayActor, statusDispatch)
- }
-
- protected def createActorSystem(actorSystemName: String): ActorSystem = {
- logger.info("Initializing internal actor system")
- ActorSystem(actorSystemName)
- }
-
- protected def createActorLoader(actorSystem: ActorSystem): ActorLoader = {
- logger.debug("Creating Simple Actor Loader")
- SimpleActorLoader(actorSystem)
- }
-
- /**
- * Does minimal setup in order to send the "starting" status message over
- * the IOPub socket
- */
- protected def initializeCoreActors(
- config: Config, actorSystem: ActorSystem, actorLoader: ActorLoader
- ) = {
- logger.debug("Creating kernel message relay actor")
- val kernelMessageRelayActor = actorSystem.actorOf(
- Props(
- classOf[KernelMessageRelay], actorLoader, true,
- Map(
- CommOpen.toTypeString -> MessageType.Incoming.CommOpen.toString,
- CommMsg.toTypeString -> MessageType.Incoming.CommMsg.toString,
- CommClose.toTypeString -> MessageType.Incoming.CommClose.toString
- ),
- Map(
- CommOpen.toTypeString -> MessageType.Outgoing.CommOpen.toString,
- CommMsg.toTypeString -> MessageType.Outgoing.CommMsg.toString,
- CommClose.toTypeString -> MessageType.Outgoing.CommClose.toString
- )
- ),
- name = SystemActorType.KernelMessageRelay.toString
- )
-
- logger.debug("Creating signature manager actor")
- val sigKey = config.getString("key")
- val sigScheme = config.getString("signature_scheme")
- logger.debug("Key = " + sigKey)
- logger.debug("Scheme = " + sigScheme)
- val signatureManagerActor = actorSystem.actorOf(
- Props(
- classOf[SignatureManagerActor], sigKey, sigScheme.replace("-", "")
- ),
- name = SecurityActorType.SignatureManager.toString
- )
-
- logger.debug("Creating status dispatch actor")
- val statusDispatch = actorSystem.actorOf(
- Props(classOf[StatusDispatch], actorLoader),
- name = SystemActorType.StatusDispatch.toString
- )
-
- logger.debug("Creating shutdown handler and sender actors")
- val shutdownHandler = actorSystem.actorOf(
- Props(classOf[ShutdownHandler], actorLoader),
- name = MessageType.Incoming.ShutdownRequest.toString
- )
- val shutdownSender = actorSystem.actorOf(
- Props(classOf[GenericSocketMessageHandler], actorLoader, SocketType.Control),
- name = MessageType.Outgoing.ShutdownReply.toString
- )
-
- (kernelMessageRelayActor, signatureManagerActor, statusDispatch, shutdownHandler, shutdownSender)
- }
-
- protected def createSockets(
- config: Config, actorSystem: ActorSystem, actorLoader: ActorLoader
- ): Unit = {
- logger.debug("Creating sockets")
-
- val socketConfig: SocketConfig = SocketConfig.fromConfig(config)
- logger.info("Connection Profile: "
- + Json.prettyPrint(Json.toJson(socketConfig)))
-
- logger.debug("Constructing ServerSocketFactory")
- val socketFactory = new SocketFactory(socketConfig)
-
- logger.debug("Initializing Heartbeat on port " +
- socketConfig.hb_port)
- val heartbeatActor = actorSystem.actorOf(
- Props(classOf[Heartbeat], socketFactory),
- name = SocketType.Heartbeat.toString
- )
-
- logger.debug("Initializing Stdin on port " +
- socketConfig.stdin_port)
- val stdinActor = actorSystem.actorOf(
- Props(classOf[Stdin], socketFactory, actorLoader),
- name = SocketType.StdIn.toString
- )
-
- logger.debug("Initializing Shell on port " +
- socketConfig.shell_port)
- val shellActor = actorSystem.actorOf(
- Props(classOf[Shell], socketFactory, actorLoader),
- name = SocketType.Shell.toString
- )
-
- logger.debug("Initializing Control on port " +
- socketConfig.control_port)
- val controlActor = actorSystem.actorOf(
- Props(classOf[Control], socketFactory, actorLoader),
- name = SocketType.Control.toString
- )
-
- logger.debug("Initializing IOPub on port " +
- socketConfig.iopub_port)
- val ioPubActor = actorSystem.actorOf(
- Props(classOf[IOPub], socketFactory),
- name = SocketType.IOPub.toString
- )
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel/src/main/scala/com/ibm/spark/boot/layer/ComponentInitialization.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/boot/layer/ComponentInitialization.scala b/kernel/src/main/scala/com/ibm/spark/boot/layer/ComponentInitialization.scala
deleted file mode 100644
index 939b896..0000000
--- a/kernel/src/main/scala/com/ibm/spark/boot/layer/ComponentInitialization.scala
+++ /dev/null
@@ -1,200 +0,0 @@
-/*
- * Copyright 2015 IBM Corp.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.ibm.spark.boot.layer
-
-import java.util
-import java.util.concurrent.ConcurrentHashMap
-
-import akka.actor.ActorRef
-import com.ibm.spark.comm.{CommManager, KernelCommManager, CommRegistrar, CommStorage}
-import com.ibm.spark.dependencies.{DependencyDownloader, IvyDependencyDownloader}
-import com.ibm.spark.global
-import com.ibm.spark.interpreter._
-import com.ibm.spark.kernel.api.{KernelLike, Kernel}
-import com.ibm.spark.kernel.protocol.v5.KMBuilder
-import com.ibm.spark.kernel.protocol.v5.kernel.ActorLoader
-import com.ibm.spark.kernel.protocol.v5.stream.KernelOutputStream
-import com.ibm.spark.magic.MagicLoader
-import com.ibm.spark.magic.builtin.BuiltinLoader
-import com.ibm.spark.magic.dependencies.DependencyMap
-import com.ibm.spark.utils.{MultiClassLoader, TaskManager, KeyValuePairUtils, LogLike}
-import com.typesafe.config.Config
-import org.apache.spark.sql.SQLContext
-import org.apache.spark.{SparkContext, SparkConf}
-
-import scala.collection.JavaConverters._
-
-import scala.util.Try
-
-/**
- * Represents the component initialization. All component-related pieces of the
- * kernel (non-actors) should be created here. Limited items should be exposed.
- */
-trait ComponentInitialization {
- /**
- * Initializes and registers all components (not needed by bare init).
- *
- * @param config The config used for initialization
- * @param appName The name of the "application" for Spark
- * @param actorLoader The actor loader to use for some initialization
- */
- def initializeComponents(
- config: Config, appName: String, actorLoader: ActorLoader
- ): (CommStorage, CommRegistrar, CommManager, Interpreter,
- Kernel, DependencyDownloader, MagicLoader,
- collection.mutable.Map[String, ActorRef])
-}
-
-/**
- * Represents the standard implementation of ComponentInitialization.
- */
-trait StandardComponentInitialization extends ComponentInitialization {
- this: LogLike =>
-
- /**
- * Initializes and registers all components (not needed by bare init).
- *
- * @param config The config used for initialization
- * @param appName The name of the "application" for Spark
- * @param actorLoader The actor loader to use for some initialization
- */
- def initializeComponents(
- config: Config, appName: String, actorLoader: ActorLoader
- ) = {
- val (commStorage, commRegistrar, commManager) =
- initializeCommObjects(actorLoader)
-
- val manager = InterpreterManager(config)
- val scalaInterpreter = manager.interpreters.get("Scala").orNull
-
- val dependencyDownloader = initializeDependencyDownloader(config)
- val magicLoader = initializeMagicLoader(
- config, scalaInterpreter, dependencyDownloader)
-
- val kernel = initializeKernel(
- config, actorLoader, manager, commManager, magicLoader
- )
-
- val responseMap = initializeResponseMap()
-
- initializeSparkContext(config, kernel, appName)
-
- (commStorage, commRegistrar, commManager,
- manager.defaultInterpreter.orNull, kernel,
- dependencyDownloader, magicLoader, responseMap)
-
- }
-
-
- def initializeSparkContext(config:Config, kernel:Kernel, appName:String) = {
- if(!config.getBoolean("nosparkcontext")) {
- kernel.createSparkContext(config.getString("spark.master"), appName)
- }
- }
-
- private def initializeCommObjects(actorLoader: ActorLoader) = {
- logger.debug("Constructing Comm storage")
- val commStorage = new CommStorage()
-
- logger.debug("Constructing Comm registrar")
- val commRegistrar = new CommRegistrar(commStorage)
-
- logger.debug("Constructing Comm manager")
- val commManager = new KernelCommManager(
- actorLoader, KMBuilder(), commRegistrar)
-
- (commStorage, commRegistrar, commManager)
- }
-
- private def initializeDependencyDownloader(config: Config) = {
- val dependencyDownloader = new IvyDependencyDownloader(
- "http://repo1.maven.org/maven2/", config.getString("ivy_local")
- )
-
- dependencyDownloader
- }
-
- protected def initializeResponseMap(): collection.mutable.Map[String, ActorRef] =
- new ConcurrentHashMap[String, ActorRef]().asScala
-
- private def initializeKernel(
- config: Config,
- actorLoader: ActorLoader,
- interpreterManager: InterpreterManager,
- commManager: CommManager,
- magicLoader: MagicLoader
- ) = {
- val kernel = new Kernel(
- config,
- actorLoader,
- interpreterManager,
- commManager,
- magicLoader
- )
- /*
- interpreter.doQuietly {
- interpreter.bind(
- "kernel", "com.ibm.spark.kernel.api.Kernel",
- kernel, List( """@transient implicit""")
- )
- }
- */
- magicLoader.dependencyMap.setKernel(kernel)
-
- kernel
- }
-
- private def initializeMagicLoader(
- config: Config, interpreter: Interpreter,
- dependencyDownloader: DependencyDownloader
- ) = {
- logger.debug("Constructing magic loader")
-
- logger.debug("Building dependency map")
- val dependencyMap = new DependencyMap()
- .setInterpreter(interpreter)
- .setKernelInterpreter(interpreter) // This is deprecated
- .setDependencyDownloader(dependencyDownloader)
- .setConfig(config)
-
- logger.debug("Creating BuiltinLoader")
- val builtinLoader = new BuiltinLoader()
-
- val magicUrlArray = config.getStringList("magic_urls").asScala
- .map(s => new java.net.URL(s)).toArray
-
- if (magicUrlArray.isEmpty)
- logger.warn("No external magics provided to MagicLoader!")
- else
- logger.info("Using magics from the following locations: " +
- magicUrlArray.map(_.getPath).mkString(","))
-
- val multiClassLoader = new MultiClassLoader(
- builtinLoader,
- interpreter.classLoader
- )
-
- logger.debug("Creating MagicLoader")
- val magicLoader = new MagicLoader(
- dependencyMap = dependencyMap,
- urls = magicUrlArray,
- parentLoader = multiClassLoader
- )
- magicLoader.dependencyMap.setMagicLoader(magicLoader)
- magicLoader
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel/src/main/scala/com/ibm/spark/boot/layer/HandlerInitialization.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/boot/layer/HandlerInitialization.scala b/kernel/src/main/scala/com/ibm/spark/boot/layer/HandlerInitialization.scala
deleted file mode 100644
index 1d16ac4..0000000
--- a/kernel/src/main/scala/com/ibm/spark/boot/layer/HandlerInitialization.scala
+++ /dev/null
@@ -1,185 +0,0 @@
-/*
- * Copyright 2014 IBM Corp.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.ibm.spark.boot.layer
-
-import akka.actor.{ActorRef, ActorSystem, Props}
-import com.ibm.spark.comm.{CommRegistrar, CommStorage}
-import com.ibm.spark.interpreter.Interpreter
-import com.ibm.spark.kernel.api.Kernel
-import com.ibm.spark.kernel.protocol.v5.MessageType.MessageType
-import com.ibm.spark.kernel.protocol.v5.SocketType.SocketType
-import com.ibm.spark.kernel.protocol.v5.handler._
-import com.ibm.spark.kernel.protocol.v5.interpreter.InterpreterActor
-import com.ibm.spark.kernel.protocol.v5.interpreter.tasks.InterpreterTaskFactory
-import com.ibm.spark.kernel.protocol.v5.kernel.ActorLoader
-import com.ibm.spark.kernel.protocol.v5.magic.{MagicParser, PostProcessor}
-import com.ibm.spark.kernel.protocol.v5.relay.ExecuteRequestRelay
-import com.ibm.spark.kernel.protocol.v5.{MessageType, SocketType, SystemActorType}
-import com.ibm.spark.magic.MagicLoader
-import com.ibm.spark.utils.LogLike
-
-/**
- * Represents the Akka handler initialization. All actors (not needed in bare
- * initialization) should be constructed here.
- */
-trait HandlerInitialization {
- /**
- * Initializes and registers all handlers.
- *
- * @param actorSystem The actor system needed for registration
- * @param actorLoader The actor loader needed for registration
- * @param kernel The kernel api needed for registration
- * @param interpreter The main interpreter needed for registration
- * @param magicLoader The magic loader needed for registration
- * @param commRegistrar The comm registrar needed for registration
- * @param commStorage The comm storage needed for registration
- */
- def initializeHandlers(
- actorSystem: ActorSystem, actorLoader: ActorLoader,
- kernel: Kernel,
- interpreter: Interpreter, magicLoader: MagicLoader,
- commRegistrar: CommRegistrar, commStorage: CommStorage,
- responseMap: collection.mutable.Map[String, ActorRef]
- ): Unit
-}
-
-/**
- * Represents the standard implementation of HandlerInitialization.
- */
-trait StandardHandlerInitialization extends HandlerInitialization {
- this: LogLike =>
-
- /**
- * Initializes and registers all handlers.
- *
- * @param actorSystem The actor system needed for registration
- * @param actorLoader The actor loader needed for registration
- * @param kernel The kernel api needed for registration
- * @param interpreter The main interpreter needed for registration
- * @param magicLoader The magic loader needed for registration
- * @param commRegistrar The comm registrar needed for registration
- * @param commStorage The comm storage needed for registration
- */
- def initializeHandlers(
- actorSystem: ActorSystem, actorLoader: ActorLoader,
- kernel: Kernel,
- interpreter: Interpreter, magicLoader: MagicLoader,
- commRegistrar: CommRegistrar, commStorage: CommStorage,
- responseMap: collection.mutable.Map[String, ActorRef]
- ): Unit = {
- initializeKernelHandlers(
- actorSystem, actorLoader, kernel, commRegistrar, commStorage, responseMap
- )
- initializeSystemActors(actorSystem, actorLoader, interpreter, magicLoader)
- }
-
- private def initializeSystemActors(
- actorSystem: ActorSystem, actorLoader: ActorLoader,
- interpreter: Interpreter, magicLoader: MagicLoader
- ): Unit = {
- logger.debug("Creating interpreter actor")
- val interpreterActor = actorSystem.actorOf(
- Props(classOf[InterpreterActor], new InterpreterTaskFactory(interpreter)),
- name = SystemActorType.Interpreter.toString
- )
-
- logger.debug("Creating execute request relay actor")
- val postProcessor = new PostProcessor(interpreter)
- val magicParser = new MagicParser(magicLoader)
- val executeRequestRelayActor = actorSystem.actorOf(
- Props(classOf[ExecuteRequestRelay],
- actorLoader, magicLoader, magicParser, postProcessor
- ),
- name = SystemActorType.ExecuteRequestRelay.toString
- )
- }
-
- private def initializeKernelHandlers(
- actorSystem: ActorSystem, actorLoader: ActorLoader,
- kernel: Kernel,
- commRegistrar: CommRegistrar, commStorage: CommStorage,
- responseMap: collection.mutable.Map[String, ActorRef]
- ): Unit = {
- def initializeRequestHandler[T](clazz: Class[T], messageType: MessageType, extraArguments: AnyRef*) = {
- logger.debug("Creating %s handler".format(messageType.toString))
- actorSystem.actorOf(
- Props(clazz, actorLoader +: extraArguments: _*),
- name = messageType.toString
- )
- }
-
- def initializeInputHandler[T](
- clazz: Class[T],
- messageType: MessageType
- ): Unit = {
- logger.debug("Creating %s handler".format(messageType.toString))
- actorSystem.actorOf(
- Props(clazz, actorLoader, responseMap),
- name = messageType.toString
- )
- }
-
- // TODO: Figure out how to pass variable number of arguments to actor
- def initializeCommHandler[T](clazz: Class[T], messageType: MessageType) = {
- logger.debug("Creating %s handler".format(messageType.toString))
- actorSystem.actorOf(
- Props(clazz, actorLoader, commRegistrar, commStorage),
- name = messageType.toString
- )
- }
-
- def initializeSocketHandler(socketType: SocketType, messageType: MessageType): Unit = {
- logger.debug("Creating %s to %s socket handler ".format(messageType.toString ,socketType.toString))
- actorSystem.actorOf(
- Props(classOf[GenericSocketMessageHandler], actorLoader, socketType),
- name = messageType.toString
- )
- }
-
- // These are the handlers for messages coming into the
- initializeRequestHandler(classOf[ExecuteRequestHandler],
- MessageType.Incoming.ExecuteRequest, kernel)
- initializeRequestHandler(classOf[KernelInfoRequestHandler],
- MessageType.Incoming.KernelInfoRequest)
- initializeRequestHandler(classOf[CodeCompleteHandler],
- MessageType.Incoming.CompleteRequest)
- initializeInputHandler(classOf[InputRequestReplyHandler],
- MessageType.Incoming.InputReply)
- initializeCommHandler(classOf[CommOpenHandler],
- MessageType.Incoming.CommOpen)
- initializeCommHandler(classOf[CommMsgHandler],
- MessageType.Incoming.CommMsg)
- initializeCommHandler(classOf[CommCloseHandler],
- MessageType.Incoming.CommClose)
-
- // These are handlers for messages leaving the kernel through the sockets
- initializeSocketHandler(SocketType.Shell, MessageType.Outgoing.KernelInfoReply)
- initializeSocketHandler(SocketType.Shell, MessageType.Outgoing.ExecuteReply)
- initializeSocketHandler(SocketType.Shell, MessageType.Outgoing.CompleteReply)
-
- initializeSocketHandler(SocketType.StdIn, MessageType.Outgoing.InputRequest)
-
- initializeSocketHandler(SocketType.IOPub, MessageType.Outgoing.ExecuteResult)
- initializeSocketHandler(SocketType.IOPub, MessageType.Outgoing.Stream)
- initializeSocketHandler(SocketType.IOPub, MessageType.Outgoing.ExecuteInput)
- initializeSocketHandler(SocketType.IOPub, MessageType.Outgoing.Status)
- initializeSocketHandler(SocketType.IOPub, MessageType.Outgoing.Error)
- initializeSocketHandler(SocketType.IOPub, MessageType.Outgoing.CommOpen)
- initializeSocketHandler(SocketType.IOPub, MessageType.Outgoing.CommMsg)
- initializeSocketHandler(SocketType.IOPub, MessageType.Outgoing.CommClose)
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel/src/main/scala/com/ibm/spark/boot/layer/HookInitialization.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/boot/layer/HookInitialization.scala b/kernel/src/main/scala/com/ibm/spark/boot/layer/HookInitialization.scala
deleted file mode 100644
index c590394..0000000
--- a/kernel/src/main/scala/com/ibm/spark/boot/layer/HookInitialization.scala
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- * Copyright 2014 IBM Corp.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.ibm.spark.boot.layer
-
-import com.ibm.spark.boot.KernelBootstrap
-import com.ibm.spark.interpreter.Interpreter
-import com.ibm.spark.utils.LogLike
-
-/**
- * Represents the hook (interrupt/shutdown) initialization. All JVM-related
- * hooks should be constructed here.
- */
-trait HookInitialization {
- /**
- * Initializes and registers all hooks except shutdown.
- *
- * @param interpreter The main interpreter
- */
- def initializeHooks(interpreter: Interpreter): Unit
-
- /**
- * Initializes the shutdown hook.
- */
- def initializeShutdownHook(): Unit
-}
-
-/**
- * Represents the standard implementation of HookInitialization.
- */
-trait StandardHookInitialization extends HookInitialization {
- this: KernelBootstrap with LogLike =>
-
- /**
- * Initializes and registers all hooks.
- *
- * @param interpreter The main interpreter
- */
- def initializeHooks(interpreter: Interpreter): Unit = {
- registerInterruptHook(interpreter)
- }
-
- /**
- * Initializes the shutdown hook.
- */
- def initializeShutdownHook(): Unit = {
- registerShutdownHook()
- }
-
- private def registerInterruptHook(interpreter: Interpreter): Unit = {
- val self = this
-
- import sun.misc.{Signal, SignalHandler}
-
- // TODO: Signals are not a good way to handle this since JVM only has the
- // proprietary sun API that is not necessarily available on all platforms
- Signal.handle(new Signal("INT"), new SignalHandler() {
- private val MaxSignalTime: Long = 3000 // 3 seconds
- var lastSignalReceived: Long = 0
-
- def handle(sig: Signal) = {
- val currentTime = System.currentTimeMillis()
- if (currentTime - lastSignalReceived > MaxSignalTime) {
- logger.info("Resetting code execution!")
- interpreter.interrupt()
-
- // TODO: Cancel group representing current code execution
- //sparkContext.cancelJobGroup()
-
- logger.info("Enter Ctrl-C twice to shutdown!")
- lastSignalReceived = currentTime
- } else {
- logger.info("Shutting down kernel")
- self.shutdown()
- }
- }
- })
- }
-
- private def registerShutdownHook(): Unit = {
- logger.debug("Registering shutdown hook")
- val self = this
- val mainThread = Thread.currentThread()
- Runtime.getRuntime.addShutdownHook(new Thread() {
- override def run() = {
- logger.info("Shutting down kernel")
- self.shutdown()
- // TODO: Check if you can magically access the spark context to stop it
- // TODO: inside a different thread
- if (mainThread.isAlive) mainThread.join()
- }
- })
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel/src/main/scala/com/ibm/spark/boot/layer/InterpreterManager.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/boot/layer/InterpreterManager.scala b/kernel/src/main/scala/com/ibm/spark/boot/layer/InterpreterManager.scala
deleted file mode 100644
index 520d68c..0000000
--- a/kernel/src/main/scala/com/ibm/spark/boot/layer/InterpreterManager.scala
+++ /dev/null
@@ -1,66 +0,0 @@
-package com.ibm.spark.boot.layer
-
-import com.ibm.spark.kernel.api.KernelLike
-import com.typesafe.config.Config
-import com.ibm.spark.interpreter._
-import scala.collection.JavaConverters._
-
-import org.slf4j.LoggerFactory
-
-case class InterpreterManager(
- default: String = "Scala",
- interpreters: Map[String, Interpreter] = Map[String, Interpreter]()
-) {
-
-
- def initializeInterpreters(kernel: KernelLike): Unit = {
- interpreters.values.foreach(interpreter =>
- interpreter.init(kernel)
- )
- }
-
- def addInterpreter(
- name:String,
- interpreter: Interpreter
- ): InterpreterManager = {
- copy(interpreters = interpreters + (name -> interpreter))
- }
-
- def defaultInterpreter: Option[Interpreter] = {
- interpreters.get(default)
- }
-}
-
-object InterpreterManager {
-
- protected val logger = LoggerFactory.getLogger(this.getClass.getName)
-
- def apply(config: Config): InterpreterManager = {
- val ip = config.getStringList("interpreter_plugins").asScala ++
- config.getStringList("default_interpreter_plugin").asScala
-
- val m = ip.foldLeft(Map[String, Interpreter]())( (acc, v) => {
- v.split(":") match {
- case Array(name, className) =>
- try {
- val i = Class
- .forName(className)
- .newInstance()
- .asInstanceOf[Interpreter]
- acc + (name -> i)
- }
- catch {
- case e:Throwable =>
- logger.error("Error loading interpreter class " + className)
- logger.error(e.getMessage())
- acc
- }
- case _ => acc
- }
- })
-
- val default = config.getString("default_interpreter")
-
- InterpreterManager(interpreters = m, default = default)
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel/src/main/scala/com/ibm/spark/comm/KernelCommManager.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/comm/KernelCommManager.scala b/kernel/src/main/scala/com/ibm/spark/comm/KernelCommManager.scala
deleted file mode 100644
index 9706ce7..0000000
--- a/kernel/src/main/scala/com/ibm/spark/comm/KernelCommManager.scala
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * Copyright 2014 IBM Corp.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.ibm.spark.comm
-
-import com.ibm.spark.annotations.Experimental
-import com.ibm.spark.kernel.protocol.v5.kernel.ActorLoader
-import com.ibm.spark.kernel.protocol.v5.{KMBuilder, UUID}
-
-/**
- * Represents a CommManager that uses a KernelCommWriter for its underlying
- * open implementation.
- *
- * @param actorLoader The actor loader to use with the ClientCommWriter
- * @param kmBuilder The KMBuilder to use with the ClientCommWriter
- * @param commRegistrar The registrar to use for callback registration
- */
-@Experimental
-class KernelCommManager(
- private val actorLoader: ActorLoader,
- private val kmBuilder: KMBuilder,
- private val commRegistrar: CommRegistrar
-) extends CommManager(commRegistrar)
-{
- /**
- * Creates a new CommWriter instance given the Comm id.
- *
- * @param commId The Comm id to use with the Comm writer
- *
- * @return The new CommWriter instance
- */
- override protected def newCommWriter(commId: UUID): CommWriter =
- new KernelCommWriter(actorLoader, kmBuilder, commId)
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel/src/main/scala/com/ibm/spark/comm/KernelCommWriter.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/comm/KernelCommWriter.scala b/kernel/src/main/scala/com/ibm/spark/comm/KernelCommWriter.scala
deleted file mode 100644
index e494663..0000000
--- a/kernel/src/main/scala/com/ibm/spark/comm/KernelCommWriter.scala
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * Copyright 2014 IBM Corp.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.ibm.spark.comm
-
-import com.ibm.spark.annotations.Experimental
-import com.ibm.spark.kernel.protocol.v5
-import com.ibm.spark.kernel.protocol.v5.content.{CommMsg, CommOpen, CommClose, CommContent}
-import com.ibm.spark.kernel.protocol.v5._
-import com.ibm.spark.kernel.protocol.v5.kernel.ActorLoader
-
-/**
- * Represents a CommWriter to send messages from the Kernel to the Client.
- *
- * @param actorLoader The actor loader to use for loading actors responsible for
- * communication
- * @param kmBuilder The kernel message builder used to construct kernel messages
- * @param commId The comm id associated with this writer (defaults to a
- * random UUID)
- */
-@Experimental
-class KernelCommWriter(
- private val actorLoader: ActorLoader,
- private val kmBuilder: KMBuilder,
- override private[comm] val commId: v5.UUID
-) extends CommWriter(commId) {
- /**
- * Sends the comm message (open/msg/close) to the actor responsible for
- * relaying messages.
- *
- * @param commContent The message to relay (will be packaged)
- *
- * @tparam T Either CommOpen, CommMsg, or CommClose
- */
- override protected[comm] def sendCommKernelMessage[
- T <: KernelMessageContent with CommContent
- ](commContent: T): Unit = {
- val messageType = commContent match {
- case _: CommOpen => CommOpen.toTypeString
- case _: CommMsg => CommMsg.toTypeString
- case _: CommClose => CommClose.toTypeString
- case _ => throw new Throwable("Invalid kernel message type!")
- }
- actorLoader.load(SystemActorType.KernelMessageRelay) !
- kmBuilder.withHeader(messageType).withContentString(commContent).build
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel/src/main/scala/com/ibm/spark/global/ExecuteRequestState.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/global/ExecuteRequestState.scala b/kernel/src/main/scala/com/ibm/spark/global/ExecuteRequestState.scala
deleted file mode 100644
index 389cf82..0000000
--- a/kernel/src/main/scala/com/ibm/spark/global/ExecuteRequestState.scala
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Copyright 2015 IBM Corp.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.ibm.spark.global
-
-import com.ibm.spark.kernel.api.Kernel
-import com.ibm.spark.kernel.protocol.v5.KernelMessage
-
-/**
- * Represents the state of the kernel messages being received containing
- * execute requests.
- */
-object ExecuteRequestState {
- private var _lastKernelMessage: Option[KernelMessage] = None
-
- /**
- * Processes the incoming kernel message and updates any associated state.
- *
- * @param kernelMessage The kernel message to process
- */
- def processIncomingKernelMessage(kernelMessage: KernelMessage) =
- _lastKernelMessage = Some(kernelMessage)
-
- /**
- * Returns the last kernel message funneled through the KernelMessageRelay
- * if any.
- *
- * @return Some KernelMessage instance if the relay has processed one,
- * otherwise None
- */
- def lastKernelMessage: Option[KernelMessage] = _lastKernelMessage
-
- /**
- * Resets the state of the ExecuteRequestState to the default.
- */
- def reset() = _lastKernelMessage = None
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel/src/main/scala/com/ibm/spark/global/ExecutionCounter.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/global/ExecutionCounter.scala b/kernel/src/main/scala/com/ibm/spark/global/ExecutionCounter.scala
deleted file mode 100644
index c00f937..0000000
--- a/kernel/src/main/scala/com/ibm/spark/global/ExecutionCounter.scala
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * Copyright 2014 IBM Corp.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.ibm.spark.global
-
-import com.ibm.spark.kernel.protocol.v5.UUID
-
-import scala.collection.concurrent._
-
-/**
- * A class to keep track of execution counts for sessions.
- */
-object ExecutionCounter {
- private val executionCounts: Map[UUID, Int] = TrieMap[UUID, Int]()
-
- /**
- * This function will increase, by 1, an integer value associated with the given key. The incremented value
- * will be returned. If the key has no value associated value, 1 will be returned.
- * @param key The key for incrementing the value.
- * @return The incremented value
- */
- def incr(key: UUID): Int = {
- (executionCounts += (key -> (executionCounts.getOrElse(key, 0) + 1))) (key)
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel/src/main/scala/com/ibm/spark/global/ScheduledTaskManager.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/global/ScheduledTaskManager.scala b/kernel/src/main/scala/com/ibm/spark/global/ScheduledTaskManager.scala
deleted file mode 100644
index 83d2cea..0000000
--- a/kernel/src/main/scala/com/ibm/spark/global/ScheduledTaskManager.scala
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * Copyright 2015 IBM Corp.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.ibm.spark.global
-
-import com.ibm.spark.utils
-
-object ScheduledTaskManager {
- lazy val instance = new utils.ScheduledTaskManager
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel/src/main/scala/com/ibm/spark/kernel/api/FactoryMethods.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/kernel/api/FactoryMethods.scala b/kernel/src/main/scala/com/ibm/spark/kernel/api/FactoryMethods.scala
deleted file mode 100644
index 36c5c5c..0000000
--- a/kernel/src/main/scala/com/ibm/spark/kernel/api/FactoryMethods.scala
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
- * Copyright 2015 IBM Corp.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.ibm.spark.kernel.api
-
-import java.io.{InputStream, OutputStream}
-
-import com.ibm.spark.kernel.protocol.v5
-import com.ibm.spark.kernel.protocol.v5.{KMBuilder, KernelMessage}
-import com.ibm.spark.kernel.protocol.v5.kernel.ActorLoader
-import com.ibm.spark.kernel.protocol.v5.stream.{KernelOutputStream, KernelInputStream}
-import com.typesafe.config.Config
-
-/**
- * Represents the methods available to stream data from the kernel to the
- * client.
- *
- * @param config The kernel configuration to use during object creation
- * @param actorLoader The actor loader to use when retrieve actors needed for
- * object creation
- * @param parentMessage The parent message to use when needed for object
- * creation
- * @param kernelMessageBuilder The builder to use when constructing kernel
- * messages inside objects created
- */
-class FactoryMethods(
- private val config: Config,
- private val actorLoader: ActorLoader,
- private val parentMessage: KernelMessage,
- private val kernelMessageBuilder: KMBuilder
-) extends FactoryMethodsLike {
- require(parentMessage != null, "Parent message cannot be null!")
-
- private[api] val kmBuilder = kernelMessageBuilder.withParent(parentMessage)
-
- /**
- * Creates a new kernel input stream.
- *
- * @param prompt The text to use as a prompt
- * @param password If true, should treat input as a password field
- *
- * @return The new KernelInputStream instance
- */
- override def newKernelInputStream(
- prompt: String = KernelInputStream.DefaultPrompt,
- password: Boolean = KernelInputStream.DefaultPassword
- ): InputStream = {
- new KernelInputStream(
- actorLoader,
- kmBuilder.withIds(parentMessage.ids),
- prompt = prompt,
- password = password
- )
- }
-
- /**
- * Creates a new kernel output stream.
- *
- * @param streamType The type of output stream (stdout/stderr)
- * @param sendEmptyOutput If true, will send message even if output is empty
- *
- * @return The new KernelOutputStream instance
- */
- override def newKernelOutputStream(
- streamType: String = KernelOutputStream.DefaultStreamType,
- sendEmptyOutput: Boolean = config.getBoolean("send_empty_output")
- ): OutputStream = {
- new v5.stream.KernelOutputStream(
- actorLoader,
- kmBuilder,
- com.ibm.spark.global.ScheduledTaskManager.instance,
- streamType = streamType,
- sendEmptyOutput = sendEmptyOutput
- )
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/68f7ddd6/kernel/src/main/scala/com/ibm/spark/kernel/api/Kernel.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/kernel/api/Kernel.scala b/kernel/src/main/scala/com/ibm/spark/kernel/api/Kernel.scala
deleted file mode 100644
index 219804a..0000000
--- a/kernel/src/main/scala/com/ibm/spark/kernel/api/Kernel.scala
+++ /dev/null
@@ -1,448 +0,0 @@
-/*
- * Copyright 2015 IBM Corp.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.ibm.spark.kernel.api
-
-import java.io.{OutputStream, InputStream, PrintStream}
-import java.util.concurrent.ConcurrentHashMap
-
-import com.ibm.spark.annotations.Experimental
-import com.ibm.spark.boot.layer.InterpreterManager
-import com.ibm.spark.comm.CommManager
-import com.ibm.spark.global
-import com.ibm.spark.interpreter.Results.Result
-import com.ibm.spark.interpreter._
-import com.ibm.spark.kernel.protocol.v5
-import com.ibm.spark.kernel.protocol.v5.{KMBuilder, KernelMessage}
-import com.ibm.spark.kernel.protocol.v5.kernel.ActorLoader
-import com.ibm.spark.kernel.protocol.v5.magic.MagicParser
-import com.ibm.spark.kernel.protocol.v5.stream.{KernelOutputStream, KernelInputStream}
-import com.ibm.spark.magic.{MagicLoader, MagicExecutor}
-import com.ibm.spark.utils.{KeyValuePairUtils, LogLike}
-import com.typesafe.config.Config
-import org.apache.spark.api.java.JavaSparkContext
-import org.apache.spark.sql.SQLContext
-import org.apache.spark.{SparkContext, SparkConf}
-import scala.util.{Try, DynamicVariable}
-
-import scala.reflect.runtime.universe._
-
-import scala.language.dynamics
-import com.ibm.spark.global.ExecuteRequestState
-
-/**
- * Represents the main kernel API to be used for interaction.
- *
- * @param _config The configuration used when starting the kernel
- * @param interpreterManager The interpreter manager to expose in this instance
- * @param comm The Comm manager to expose in this instance
- * @param actorLoader The actor loader to use for message relaying
- */
-@Experimental
-class Kernel (
- private val _config: Config,
- private val actorLoader: ActorLoader,
- val interpreterManager: InterpreterManager,
- val comm: CommManager,
- val magicLoader: MagicLoader
-) extends KernelLike with LogLike {
-
- /**
- * Represents the current input stream used by the kernel for the specific
- * thread.
- */
- private val currentInputStream =
- new DynamicVariable[InputStream](null)
- private val currentInputKernelMessage =
- new DynamicVariable[KernelMessage](null)
-
- /**
- * Represents the current output stream used by the kernel for the specific
- * thread.
- */
- private val currentOutputStream =
- new DynamicVariable[PrintStream](null)
- private val currentOutputKernelMessage =
- new DynamicVariable[KernelMessage](null)
-
- /**
- * Represents the current error stream used by the kernel for the specific
- * thread.
- */
- private val currentErrorStream =
- new DynamicVariable[PrintStream](null)
- private val currentErrorKernelMessage =
- new DynamicVariable[KernelMessage](null)
-
- private var _sparkContext:SparkContext = null;
- private var _sparkConf:SparkConf = null;
- private var _javaSparkContext:JavaSparkContext = null;
- private var _sqlContext:SQLContext = null;
-
- /**
- * Represents magics available through the kernel.
- */
- val magics = new MagicExecutor(magicLoader)
-
- /**
- * Represents magic parsing functionality.
- */
- val magicParser = new MagicParser(magicLoader)
-
- /**
- * Represents the data that can be shared using the kernel as the middleman.
- *
- * @note Using Java structure to enable other languages to have easy access!
- */
- val data: java.util.Map[String, Any] = new ConcurrentHashMap[String, Any]()
-
-
- interpreterManager.initializeInterpreters(this)
-
- val interpreter = interpreterManager.defaultInterpreter.get
-
- /**
- * Handles the output of interpreting code.
- * @param output the output of the interpreter
- * @return (success, message) or (failure, message)
- */
- private def handleInterpreterOutput(
- output: (Result, Either[ExecuteOutput, ExecuteFailure])
- ): (Boolean, String) = {
- val (success, result) = output
- success match {
- case Results.Success =>
- (true, result.left.getOrElse("").asInstanceOf[String])
- case Results.Error =>
- (false, result.right.getOrElse("").toString)
- case Results.Aborted =>
- (false, "Aborted!")
- case Results.Incomplete =>
- // If we get an incomplete it's most likely a syntax error, so
- // let the user know.
- (false, "Syntax Error!")
- }
- }
-
- override def config:Config = {
- _config
- }
-
- /**
- * Executes a block of code represented as a string and returns the result.
- *
- * @param code The code as an option to execute
- * @return A tuple containing the result (true/false) and the output as a
- * string
- */
- def eval(code: Option[String]): (Boolean, String) = {
- code.map(c => {
- magicParser.parse(c) match {
- case Left(parsedCode) =>
- val output = interpreter.interpret(parsedCode)
- handleInterpreterOutput(output)
- case Right(errMsg) =>
- (false, errMsg)
- }
- }).getOrElse((false, "Error!"))
- }
-
- /**
- * Constructs a new instance of the stream methods using the latest
- * kernel message instance.
- *
- * @return The collection of stream methods
- */
- override def stream: StreamMethods = stream()
-
- /**
- * Constructs a new instance of the stream methods using the specified
- * kernel message instance.
- *
- * @param parentMessage The message to serve as the parent of outgoing
- * messages sent as a result of using streaming methods
- *
- * @return The collection of streaming methods
- */
- private[spark] def stream(
- parentMessage: v5.KernelMessage = lastKernelMessage()
- ): StreamMethods = {
- new StreamMethods(actorLoader, parentMessage)
- }
-
- /**
- * Constructs a new instance of the factory methods using the latest
- * kernel message instance.
- *
- * @return The collection of factory methods
- */
- override def factory: FactoryMethods = factory()
-
- /**
- * Constructs a new instance of the factory methods using the specified
- * kernel message and kernel message builder.
- *
- * @param parentMessage The message to serve as the parent of outgoing
- * messages sent as a result of using an object created
- * by the factory methods
- * @param kmBuilder The builder to be used by objects created by factory
- * methods
- *
- * @return The collection of factory methods
- */
- private[spark] def factory(
- parentMessage: v5.KernelMessage = lastKernelMessage(),
- kmBuilder: v5.KMBuilder = v5.KMBuilder()
- ): FactoryMethods = {
- new FactoryMethods(_config, actorLoader, parentMessage, kmBuilder)
- }
-
- /**
- * Returns a print stream to be used for communication back to clients
- * via standard out.
- *
- * @return The print stream instance or an error if the stream info is
- * not found
- */
- override def out: PrintStream = {
- val kernelMessage = lastKernelMessage()
-
- constructStream(currentOutputStream, currentOutputKernelMessage, kernelMessage, { kernelMessage =>
- val outputStream = this.factory(parentMessage = kernelMessage)
- .newKernelOutputStream("stdout")
-
- new PrintStream(outputStream)
- })
- }
-
- /**
- * Returns a print stream to be used for communication back to clients
- * via standard error.
- *
- * @return The print stream instance or an error if the stream info is
- * not found
- */
- override def err: PrintStream = {
- val kernelMessage = lastKernelMessage()
-
- constructStream(currentErrorStream, currentErrorKernelMessage, kernelMessage, { kernelMessage =>
- val outputStream = this.factory(parentMessage = kernelMessage)
- .newKernelOutputStream("stderr")
-
- new PrintStream(outputStream)
- })
- }
-
- /**
- * Returns an input stream to be used to receive information from the client.
- *
- * @return The input stream instance or an error if the stream info is
- * not found
- */
- override def in: InputStream = {
- val kernelMessage = lastKernelMessage()
-
- constructStream(currentInputStream, currentInputKernelMessage, kernelMessage, { kernelMessage =>
- this.factory(parentMessage = kernelMessage).newKernelInputStream()
- })
- }
-
- /**
- * Constructs or uses an existing stream.
- *
- * @param dynamicStream The DynamicVariable containing the stream to modify
- * or use
- * @param dynamicKernelMessage The DynamicVariable containing the KernelMessage to
- * check against the new KernelMessage
- * @param newKernelMessage The potentially-new KernelMessage
- * @param streamConstructionFunc The function used to create a new stream
- * @param typeTag The type information associated with the stream
- * @tparam T The stream type
- * @return The new stream or existing stream
- */
- private def constructStream[T](
- dynamicStream: DynamicVariable[T],
- dynamicKernelMessage: DynamicVariable[KernelMessage],
- newKernelMessage: KernelMessage,
- streamConstructionFunc: (KernelMessage) => T
- )(implicit typeTag: TypeTag[T]) = {
- // Update the stream being used only if the information has changed
- // or if the stream has not been initialized
- if (updateKernelMessage(dynamicKernelMessage, newKernelMessage) ||
- dynamicStream.value == null)
- {
- logger.trace("Creating new kernel " + typeTag.tpe.toString + "!")
- dynamicStream.value = streamConstructionFunc(newKernelMessage)
- }
-
- dynamicStream.value
- }
-
- /**
- * Updates the last stream info returning the status of whether or not the
- * new stream info was different than the last stream info.
- *
- * @param dynamicKernelMessage The dynamic variable containing the current
- * stream info
- * @param kernelMessage The new stream info
- * @return True if the new stream info is different from the last (therefore
- * replaced), otherwise false
- */
- private def updateKernelMessage(
- dynamicKernelMessage: DynamicVariable[KernelMessage],
- kernelMessage: KernelMessage
- ): Boolean =
- if (kernelMessage != null && !kernelMessage.equals(dynamicKernelMessage.value)) {
- dynamicKernelMessage.value = kernelMessage
- true
- } else {
- false
- }
-
- /**
- * Retrieves the last kernel message received by the kernel.
- *
- * @throws IllegalArgumentException If no kernel message has been received
- *
- * @return The kernel message instance
- */
- private def lastKernelMessage() = {
- val someKernelMessage = ExecuteRequestState.lastKernelMessage
- require(someKernelMessage.nonEmpty, "No kernel message received!")
- someKernelMessage.get
- }
-
- override def createSparkContext(conf: SparkConf): SparkContext = {
- _sparkConf = createSparkConf(conf)
- _sparkContext = initializeSparkContext(sparkConf)
- _javaSparkContext = new JavaSparkContext(_sparkContext)
- _sqlContext = initializeSqlContext(_sparkContext)
-
- val sparkMaster = _sparkConf.getOption("spark.master").getOrElse("not_set")
- logger.info( s"Connecting to spark.master $sparkMaster")
-
- updateInterpreterWithSparkContext(interpreter, sparkContext)
- updateInterpreterWithSqlContext(interpreter, sqlContext)
-
- magicLoader.dependencyMap =
- magicLoader.dependencyMap.setSparkContext(_sparkContext)
-
- _sparkContext
- }
-
- override def createSparkContext(
- master: String, appName: String
- ): SparkContext = {
- createSparkContext(new SparkConf().setMaster(master).setAppName(appName))
- }
-
- // TODO: Think of a better way to test without exposing this
- protected[kernel] def createSparkConf(conf: SparkConf) = {
-
- logger.info("Setting deployMode to client")
- conf.set("spark.submit.deployMode", "client")
-
- KeyValuePairUtils.stringToKeyValuePairSeq(
- _config.getString("spark_configuration")
- ).foreach { keyValuePair =>
- logger.info(s"Setting ${keyValuePair.key} to ${keyValuePair.value}")
- Try(conf.set(keyValuePair.key, keyValuePair.value))
- }
-
- // TODO: Move SparkIMain to private and insert in a different way
- logger.warn("Locked to Scala interpreter with SparkIMain until decoupled!")
-
- // TODO: Construct class server outside of SparkIMain
- logger.warn("Unable to control initialization of REPL class server!")
- logger.info("REPL Class Server Uri: " + interpreter.classServerURI)
- conf.set("spark.repl.class.uri", interpreter.classServerURI)
-
- conf
- }
-
- // TODO: Think of a better way to test without exposing this
- protected[kernel] def initializeSparkContext(sparkConf: SparkConf): SparkContext = {
-
- logger.debug("Constructing new Spark Context")
- // TODO: Inject stream redirect headers in Spark dynamically
- var sparkContext: SparkContext = null
- val outStream = new KernelOutputStream(
- actorLoader, KMBuilder(), global.ScheduledTaskManager.instance,
- sendEmptyOutput = _config.getBoolean("send_empty_output")
- )
-
- // Update global stream state and use it to set the Console local variables
- // for threads in the Spark threadpool
- global.StreamState.setStreams(System.in, outStream, outStream)
- global.StreamState.withStreams {
- sparkContext = new SparkContext(sparkConf)
- }
-
- sparkContext
- }
-
- // TODO: Think of a better way to test without exposing this
- protected[kernel] def updateInterpreterWithSparkContext(
- interpreter: Interpreter, sparkContext: SparkContext
- ) = {
-
- interpreter.bindSparkContext(sparkContext)
- }
-
- protected[kernel] def initializeSqlContext(
- sparkContext: SparkContext
- ): SQLContext = {
- val sqlContext: SQLContext = try {
- logger.info("Attempting to create Hive Context")
- val hiveContextClassString =
- "org.apache.spark.sql.hive.HiveContext"
-
- logger.debug(s"Looking up $hiveContextClassString")
- val hiveContextClass = Class.forName(hiveContextClassString)
-
- val sparkContextClass = classOf[SparkContext]
- val sparkContextClassName = sparkContextClass.getName
-
- logger.debug(s"Searching for constructor taking $sparkContextClassName")
- val hiveContextContructor =
- hiveContextClass.getConstructor(sparkContextClass)
-
- logger.debug("Invoking Hive Context constructor")
- hiveContextContructor.newInstance(sparkContext).asInstanceOf[SQLContext]
- } catch {
- case _: Throwable =>
- logger.warn("Unable to create Hive Context! Defaulting to SQL Context!")
- new SQLContext(sparkContext)
- }
-
- sqlContext
- }
-
- protected[kernel] def updateInterpreterWithSqlContext(
- interpreter: Interpreter, sqlContext: SQLContext
- ): Unit = {
- interpreter.bindSqlContext(sqlContext)
- }
-
- override def interpreter(name: String): Option[Interpreter] = {
- interpreterManager.interpreters.get(name)
- }
-
- override def sparkContext: SparkContext = _sparkContext
- override def sparkConf: SparkConf = _sparkConf
- override def javaSparkContext: JavaSparkContext = _javaSparkContext
- override def sqlContext: SQLContext = _sqlContext
-}