diff --git a/kernel-api/src/main/scala/com/ibm/spark/magic/dependencies/DependencyMap.scala b/kernel-api/src/main/scala/com/ibm/spark/magic/dependencies/DependencyMap.scala
index e8db370..e91a2f2 100644
--- a/kernel-api/src/main/scala/com/ibm/spark/magic/dependencies/DependencyMap.scala
+++ b/kernel-api/src/main/scala/com/ibm/spark/magic/dependencies/DependencyMap.scala
@@ -53,6 +53,7 @@
    * Sets the Interpreter for this map.
    * @param interpreter The new Interpreter
    */
+  //@deprecated("Use setInterpreter with IncludeInterpreter!", "2015.05.06")
   def setKernelInterpreter(interpreter: Interpreter) = {
     internalMap(typeOf[IncludeKernelInterpreter]) =
       PartialFunction[Magic, Unit](
@@ -123,7 +124,7 @@
 
   /**
    * Sets the MagicLoader for this map.
-   * @param kernel The new Kernel
+   * @param magicLoader The new MagicLoader
    */
   def setMagicLoader(magicLoader: MagicLoader) = {
     internalMap(typeOf[IncludeMagicLoader]) =
diff --git a/kernel-api/src/main/scala/com/ibm/spark/magic/dependencies/IncludeKernelInterpreter.scala b/kernel-api/src/main/scala/com/ibm/spark/magic/dependencies/IncludeKernelInterpreter.scala
index 60faf50..de19c07 100644
--- a/kernel-api/src/main/scala/com/ibm/spark/magic/dependencies/IncludeKernelInterpreter.scala
+++ b/kernel-api/src/main/scala/com/ibm/spark/magic/dependencies/IncludeKernelInterpreter.scala
@@ -19,6 +19,7 @@
 import com.ibm.spark.interpreter.Interpreter
 import com.ibm.spark.magic.Magic
 
+//@deprecated("Use IncludeInterpreter instead!", "2015.05.06")
 trait IncludeKernelInterpreter {
   this: Magic =>
 
diff --git a/kernel-api/src/main/scala/com/ibm/spark/utils/TaskManager.scala b/kernel-api/src/main/scala/com/ibm/spark/utils/TaskManager.scala
index 55b1b97..c3a43fc 100644
--- a/kernel-api/src/main/scala/com/ibm/spark/utils/TaskManager.scala
+++ b/kernel-api/src/main/scala/com/ibm/spark/utils/TaskManager.scala
@@ -1,93 +1,104 @@
-/*
- * 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.UUID
-import java.util.concurrent.{TimeUnit, ArrayBlockingQueue}
-import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.atomic.AtomicInteger
 
-import scala.concurrent.{Future, Promise, promise}
+import org.slf4j.LoggerFactory
+
+import scala.concurrent.{promise, Future}
+import java.util.concurrent._
 
 import com.ibm.spark.security.KernelSecurityManager._
+import TaskManager._
+
+import scala.util.Try
 
 /**
- * Represents a generic manager of Runnable tasks that will be executed in a
- * separate thread (created inside the manager). Facilitates running tasks and
- * provides a method to kill
+ * Represents a processor of tasks that has X worker threads dedicated to
+ * executing the tasks.
+ *
+ * @param threadGroup The thread group to use with all worker threads
+ * @param minimumWorkers The number of workers to spawn initially and keep
+ *                       alive even when idle
+ * @param maximumWorkers The max number of worker threads to spawn, defaulting
+ *                   to the number of processors on the machine
+ * @param keepAliveTime The maximum time in milliseconds for workers to remain
+ *                      idle before shutting down
  */
 class TaskManager(
-  taskThreadGroup: ThreadGroup = new ThreadGroup(RestrictedGroupName),
-  taskThreadName: String = "task-" + UUID.randomUUID().toString
+  private val threadGroup: ThreadGroup = DefaultThreadGroup,
+  private val maxTasks: Int = DefaultMaxTasks,
+  private val minimumWorkers: Int = DefaultMinimumWorkers,
+  private val maximumWorkers: Int = DefaultMaximumWorkers,
+  private val keepAliveTime: Long = DefaultKeepAliveTime
 ) {
-  // Maximum time to wait (in milliseconds) before forcefully stopping this
-  // thread when an interrupt fails
-  private val InterruptTimeout = 5 * 1000
-  private val _queueCapacity = 200
-  private val _taskQueue =
-    new ArrayBlockingQueue[(Runnable, Promise[_])](_queueCapacity)
-  private var _taskThread: TaskThread = _
+  protected val logger = LoggerFactory.getLogger(this.getClass.getName)
 
-  private val _currentPromise: AtomicReference[Promise[_]] =
-    new AtomicReference[Promise[_]]()
+  private class TaskManagerThreadFactory extends ThreadFactory {
+    override def newThread(r: Runnable): Thread = {
+      val thread = new Thread(threadGroup, r)
 
-  private class TaskThread extends Thread(taskThreadGroup, taskThreadName) {
-    private[TaskManager] var _currentTask: Runnable = _
-    private[TaskManager] var _running = false
+      logger.trace(s"Creating new thread named ${thread.getName}!")
 
-    override def start(): Unit = {
-      _running = true
-      super.start()
-    }
-
-    /**
-     * Main execution loop of the task thread.
-     *
-     * Pulls tasks from an internal queue to be processed sequentially.
-     */
-    override def run(): Unit = {
-      while (_running) {
-        val element = _taskQueue.poll(1L, TimeUnit.MILLISECONDS)
-        if (element != null) {
-          _currentTask = element._1
-          _currentPromise.set(element._2)
-
-          if (_currentTask != null) _currentTask.run()
-
-          _currentTask = null
-          _currentPromise.set(null)
-        }
-      }
-    }
-
-    /**
-     * Marks the internal flag for running new tasks to false.
-     */
-    def cancel(): Unit = {
-      _running = false
+      thread
     }
   }
 
-  /**
-   * Represents the internal thread used to process tasks.
-   *
-   * @return The Thread instance wrapped in an Option, or None if not started
-   */
-  def thread: Option[Thread] =
-    if (_taskThread != null) Some(_taskThread) else None
+  private[utils] class ScalingThreadPoolExecutor extends ThreadPoolExecutor(
+    minimumWorkers,
+    maximumWorkers,
+    keepAliveTime,
+    TimeUnit.MILLISECONDS,
+    taskQueue,
+    taskManagerThreadFactory
+  ) {
+    protected val logger = LoggerFactory.getLogger(this.getClass.getName)
+
+    /** Used to keep track of tasks separately from the task queue. */
+    private val taskCount = new AtomicInteger(0)
+
+    /**
+     * Syncs the core pool size of the executor with the current number of
+     * tasks, using the minimum worker size and maximum worker size as the
+     * bounds.
+     */
+    private def syncPoolLimits(): Unit = {
+      val totalTasks = taskCount.get()
+      val newCorePoolSize =
+        math.max(minimumWorkers, math.min(totalTasks, maximumWorkers))
+
+      logger.trace(Seq(
+        s"Task execution count is $totalTasks!",
+        s"Updating core pool size to $newCorePoolSize!"
+      ).mkString(" "))
+      executor.foreach(_.setCorePoolSize(newCorePoolSize))
+    }
+
+    override def execute(r: Runnable): Unit = {
+      synchronized {
+        if (taskCount.incrementAndGet() > maximumWorkers)
+          logger.warn(s"Exceeded $maximumWorkers workers during processing!")
+
+        syncPoolLimits()
+      }
+
+      super.execute(r)
+    }
+
+    override def afterExecute(r: Runnable, t: Throwable): Unit = {
+      super.afterExecute(r, t)
+
+      synchronized {
+        taskCount.decrementAndGet()
+        syncPoolLimits()
+      }
+    }
+  }
+
+  private val taskManagerThreadFactory = new TaskManagerThreadFactory
+  private val taskQueue = new ArrayBlockingQueue[Runnable](maxTasks)
+
+  @volatile
+  private[utils] var executor: Option[ScalingThreadPoolExecutor] = None
 
   /**
    * Adds a new task to the list to execute.
@@ -97,189 +108,130 @@
    * @return Future representing the return value (or error) from the task
    */
   def add[T <: Any](taskFunction: => T): Future[T] = {
+    assert(executor.nonEmpty, "Task manager not started!")
+
     val taskPromise = promise[T]()
 
     // Construct runnable that completes the promise
-    _taskQueue.add((new Runnable {
-      override def run(): Unit =
+    logger.trace(s"Queueing new task to be processed!")
+    executor.foreach(_.execute(new Runnable {
+      override def run(): Unit = {
+        var threadName: String = "???"
         try {
+          threadName = Try(Thread.currentThread().getName).getOrElse(threadName)
+          logger.trace(s"(Thread $threadName) Executing task!")
           val result = taskFunction
+
+          logger.trace(s"(Thread $threadName) Task finished successfully!")
           taskPromise.success(result)
         } catch {
-          case ex: Throwable => taskPromise.tryFailure(ex)
+          case ex: Throwable =>
+            val exName = ex.getClass.getName
+            val exMessage = Option(ex.getLocalizedMessage).getOrElse("???")
+            logger.trace(
+              s"(Thread $threadName) Task failed: ($exName) = $exMessage")
+            taskPromise.tryFailure(ex)
         }
-    }, taskPromise))
+      }
+    }))
 
     taskPromise.future
   }
 
   /**
-   * Returns the count of tasks including the currently-running one.
+   * Returns the count of tasks including the currently-running ones.
    *
    * @return The count of tasks
    */
-  def size: Int = _taskQueue.size()
+  def size: Int = taskQueue.size() + executor.map(_.getActiveCount).getOrElse(0)
 
   /**
    * Returns whether or not there is a task in the queue to be processed.
    *
    * @return True if the internal queue is not empty, otherwise false
    */
-  def hasTaskInQueue: Boolean = !_taskQueue.isEmpty
-
-  /**
-   * Returns the current executing task.
-   *
-   * @return The current task or None if no task is running
-   */
-  def currentTask: Option[Runnable] =
-    if (_taskThread != null && _taskThread._currentTask != null)
-      Some(_taskThread._currentTask)
-    else None
-
-  /**
-   * Returns the sequence of tasks.
-   *
-   * @return The sequence of tasks as Runnables
-   */
-  def tasks: Seq[Runnable] = _taskQueue.toArray.map(_.asInstanceOf[(Runnable, _)]._1)
+  def hasTaskInQueue: Boolean = !taskQueue.isEmpty
 
   /**
    * Whether or not there is a task being executed currently.
    *
    * @return True if there is a task being executed, otherwise false
    */
-  def isExecutingTask: Boolean = currentTask.nonEmpty
-
-  /**
-   * Whether or not the task manager is processing new tasks.
-   *
-   * @return True if the manager is capable of consuming more tasks,
-   *         otherwise false
-   */
-  def isRunning: Boolean = _taskThread != null && _taskThread._running
+  def isExecutingTask: Boolean = executor.exists(_.getActiveCount > 0)
 
   /**
    * Block execution (by sleeping) until all tasks currently queued up for
    * execution are processed.
    */
   def await(): Unit =
-    while (hasTaskInQueue || isExecutingTask) Thread.sleep(1)
+    while (!taskQueue.isEmpty || isExecutingTask) Thread.sleep(1)
 
   /**
-   * Starts the task manager (begins processing tasks). Creates a new thread
+   * Starts the task manager (begins processing tasks). Creates X new threads
    * in the process.
    */
-  def start(): Unit = startTaskProcessingThread()
+  def start(): Unit = {
+    logger.trace(
+      s"""
+         |Initializing with the following settings:
+         |- $minimumWorkers core worker pool
+         |- $maximumWorkers maximum workers
+         |- $keepAliveTime milliseconds keep alive time
+       """.stripMargin.trim)
+    executor = Some(new ScalingThreadPoolExecutor)
+  }
 
   /**
    * Restarts internal processing of tasks (removing current task).
    */
-  def restart(): Unit = restartTaskProcessingThread()
+  def restart(): Unit = {
+    stop()
+    start()
+  }
 
   /**
    * Stops internal processing of tasks.
-   *
-   * @param killThread Whether to kill the thread processing tasks if not it
-   *                   is not responding to interrupts
-   * @param killTimeout The period of time (in milliseconds) to wait before
-   *                    attempting to kill the thread processing tasks
    */
-  def stop(
-    killThread: Boolean = true,
-    killTimeout: Int = InterruptTimeout
-  ): Unit = killTaskProcessingThread(killThread, killTimeout)
-
-  /**
-   * Creates a new task-processing thread and starts it.
-   */
-  private def startTaskProcessingThread(): Unit = {
-    _taskThread = new TaskThread
-    _taskThread.start()
-  }
-
-  /**
-   * Attempts to cancel/interrupt the task manager's internal task-processing
-   * thread. If unable to interrupt, will be forcefully stopped after the
-   * interrupt timeout threshold is reached.
-   *
-   * @param killThread Whether to kill the thread processing tasks if not it
-   *                   is not responding to interrupts
-   * @param killTimeout The period of time (in milliseconds) to wait before
-   *                    attempting to kill the thread processing tasks
-   */
-  private def killTaskProcessingThread(
-    killThread: Boolean = true,
-    killTimeout: Int = InterruptTimeout
-  ): Unit = {
-    // NOTE: Dirty hack to suppress deprecation warnings
-    // See https://issues.scala-lang.org/browse/SI-7934 for discussion
-    object Thread {
-      @deprecated("", "") class Killer {
-        def stop() = try { _taskThread.stop() }
-      }
-
-      object Killer extends Killer
-    }
-
-    _taskThread.cancel()
-    _taskThread.interrupt()
-
-    runIfTimeout(
-      killTimeout,
-      _taskThread.isAlive && killThread,
-      // Still available (JDK8 now calls stop0 directly)
-      //
-      // Used because of discussion on:
-      // https://issues.scala-lang.org/browse/SI-6302
-      {
-        //_taskThread.stop()
-        Thread.Killer.stop()
-        val currentPromise = _currentPromise.get()
-        if (currentPromise != null)
-          currentPromise.tryFailure(new ThreadDeath())
-      }
-    )
-
-    _taskThread = null
-  }
-
-  /**
-   * Shuts down the current thread used to execute tasks and starts a new
-   * thread in its place.
-   */
-  private def restartTaskProcessingThread(): Unit = {
-    killTaskProcessingThread()
-    startTaskProcessingThread()
-  }
-
-  /**
-   * Execute body if the millisecond timeout is reached and the condition
-   * still holds true.
-   *
-   * @param milliseconds The timeout limit in milliseconds
-   * @param condition The condition to test up until executing the body
-   * @param body The body to execute
-   *
-   * @return Whether or not the body was executed
-   */
-  private def runIfTimeout(
-    milliseconds: Int,
-    condition: => Boolean,
-    body: => Unit
-  ): Boolean = {
-    val endTime: Long = milliseconds + java.lang.System.currentTimeMillis()
-
-    while (java.lang.System.currentTimeMillis() < endTime) {
-      // Exit if our condition suddenly goes south
-      if (!condition) return false
-
-      // Allow check to be interrupted
-      if (Thread.interrupted())
-        throw new InterruptedException()
-    }
-
-    if (condition) { body; true } else false
+  def stop(): Unit = {
+    executor.foreach(_.shutdownNow())
+    executor = None
   }
 }
 
+/**
+ * Represents constants associated with the task manager.
+ */
+object TaskManager {
+  /** The default thread group to use with all worker threads. */
+  val DefaultThreadGroup = new ThreadGroup(RestrictedGroupName)
+
+  /** The default number of maximum tasks accepted by the task manager. */
+  val DefaultMaxTasks = 200
+
+  /**
+   * The default number of workers to spawn initially and keep alive
+   * even when idle.
+   */
+  val DefaultMinimumWorkers = 1
+
+  /** The default maximum number of workers to spawn. */
+  val DefaultMaximumWorkers = Runtime.getRuntime.availableProcessors()
+
+  /** The default timeout in milliseconds for workers waiting for tasks. */
+  val DefaultKeepAliveTime = 1000
+
+  /**
+   * The default timeout in milliseconds to wait before stopping a thread
+   * if it cannot be interrupted.
+   */
+  val InterruptTimeout = 5000
+
+  /** The maximum time to wait to add a task to the queue in milliseconds. */
+  val MaximumTaskQueueTimeout = 10000
+
+  /**
+   * The maximum time in milliseconds to wait to queue up a thread in the
+   * thread factory.
+   */
+  val MaximumThreadQueueTimeout = 10000
+}
\ No newline at end of file
diff --git a/kernel-api/src/test/scala/com/ibm/spark/utils/TaskManagerSpec.scala b/kernel-api/src/test/scala/com/ibm/spark/utils/TaskManagerSpec.scala
index e453dcb..a4184d4 100644
--- a/kernel-api/src/test/scala/com/ibm/spark/utils/TaskManagerSpec.scala
+++ b/kernel-api/src/test/scala/com/ibm/spark/utils/TaskManagerSpec.scala
@@ -16,10 +16,10 @@
 
 package com.ibm.spark.utils
 
-import java.util.concurrent.ExecutionException
+import java.util.concurrent.{RejectedExecutionException, ExecutionException}
 
 import org.scalatest.concurrent.PatienceConfiguration.Timeout
-import org.scalatest.concurrent.{Eventually, ScalaFutures}
+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}
@@ -30,7 +30,7 @@
 
 class TaskManagerSpec extends FunSpec with Matchers with MockitoSugar
   with BeforeAndAfter with ScalaFutures with UncaughtExceptionSuppression
-  with Eventually
+  with Eventually with Timeouts
 {
   private var taskManager: TaskManager = _
   implicit override val patienceConfig = PatienceConfig(
@@ -48,18 +48,37 @@
 
   describe("TaskManager") {
     describe("#add") {
-      // TODO: How to verify the (Runnable, Promise[_]) stored in private queue?
+      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 500) 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") {
-        val f = taskManager.add { }
-
         taskManager.start()
 
+        val f = taskManager.add { }
+
         whenReady(f) { result =>
           result shouldBe a [BoxedUnit]
           taskManager.stop()
@@ -67,11 +86,11 @@
       }
 
       it("should construct a Runnable that invokes a Promise on success") {
+        taskManager.start()
+
         val returnValue = 3
         val f = taskManager.add { returnValue }
 
-        taskManager.start()
-
         whenReady(f) { result =>
           result should be (returnValue)
           taskManager.stop()
@@ -79,38 +98,26 @@
       }
 
       it("should construct a Runnable that invokes a Promise on failure") {
+        taskManager.start()
+
         val error = new Throwable("ERROR")
         val f = taskManager.add { throw error }
 
-        taskManager.start()
-
         whenReady(f.failed) { result =>
           result should be (error)
           taskManager.stop()
         }
       }
-    }
 
-    describe("#tasks") {
-      it("should return a sequence of Runnables not yet executed") {
-        // TODO: Investigate how to validate tasks better than just a count
-        for (x <- 1 to 50) taskManager.add { }
+      it("should not block when adding more tasks than available threads") {
+        val taskManager = new TaskManager(maximumWorkers = 1)
 
-        taskManager.tasks should have size 50
-      }
-    }
-
-    describe("#thread") {
-      it("should return Some(Thread) if running") {
         taskManager.start()
 
-        taskManager.thread should not be (None)
-
-        taskManager.stop()
-      }
-
-      it("should return None if not running") {
-        taskManager.thread should be (None)
+        failAfter(Span(100, Milliseconds)) {
+          taskManager.add { while (true) { Thread.sleep(1) } }
+          taskManager.add { while (true) { Thread.sleep(1) } }
+        }
       }
     }
 
@@ -119,22 +126,27 @@
         taskManager.size should be (0)
       }
 
-      it("should be one when a new task has been added") {
-        taskManager.add {}
+      it("should reflect queued tasks and executing tasks") {
+        val taskManager = new TaskManager(maximumWorkers = 1)
+        taskManager.start()
 
-        taskManager.size should be (1)
-      }
-
-      it("should be zero when the only task is currently being executed") {
+        // 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 (0)
+        taskManager.size should be (1)
 
         taskManager.stop()
       }
@@ -145,17 +157,22 @@
         taskManager.hasTaskInQueue should be (false)
       }
 
-      it("should be true when one task has been added but not started") {
-        taskManager.add {}
+      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.add { while (true) { Thread.sleep(1000) } }
-
         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)
@@ -194,55 +211,16 @@
       }
     }
 
-    describe("#currentTask") {
-      it("should be None when there are no tasks") {
-        taskManager.currentTask should be (None)
-      }
-
-      it("should be None when there are tasks, but none are running") {
-        taskManager.add { }
-
-        taskManager.currentTask should be (None)
-      }
-
-      it("should be Some(...) when a task is being executed") {
-        taskManager.add { while (true) { Thread.sleep(1000) } }
-        taskManager.start()
-
-        // Wait until executing task
-        while (!taskManager.isExecutingTask) Thread.sleep(1)
-
-        taskManager.currentTask should not be (None)
-
-        taskManager.stop()
-      }
-    }
-
-    describe("#isRunning") {
-      it("should be false when not started") {
-        taskManager.isRunning should be (false)
-      }
-
-      it("should be true after being started") {
-        taskManager.start()
-        taskManager.isRunning should be (true)
-        taskManager.stop()
-      }
-
-      it("should be false after being stopped") {
-        taskManager.start(); taskManager.stop()
-        taskManager.isRunning should be (false)
-      }
-    }
-
     describe("#await") {
       it("should block until all tasks are completed") {
-        // TODO: Need better way to ensure tasks are still running while
-        // awaiting their return
-        for (x <- 1 to 50) taskManager.add { Thread.sleep(1) }
+        val taskManager = new TaskManager(maximumWorkers = 1, maxTasks = 500)
 
         taskManager.start()
 
+        // TODO: Need better way to ensure tasks are still running while
+        // awaiting their return
+        for (x <- 1 to 500) taskManager.add { Thread.sleep(1) }
+
         assume(taskManager.hasTaskInQueue)
         taskManager.await()
 
@@ -254,10 +232,10 @@
     }
 
     describe("#start") {
-      it("should create an internal thread and start it") {
+      it("should create an internal thread pool executor") {
         taskManager.start()
 
-        taskManager.thread should not be (None)
+        taskManager.executor should not be (None)
 
         taskManager.stop()
       }
@@ -267,11 +245,11 @@
       it("should stop & erase the old internal thread and create a new one") {
         taskManager.start()
 
-        val oldThread = taskManager.thread
+        val oldExecutor = taskManager.executor
 
         taskManager.restart()
 
-        taskManager.thread should not be (oldThread)
+        taskManager.executor should not be (oldExecutor)
 
         taskManager.stop()
       }
@@ -279,8 +257,8 @@
 
     describe("#stop") {
       it("should attempt to interrupt the currently-running task") {
-        val f = taskManager.add { while (true) { Thread.sleep(1000) } }
         taskManager.start()
+        val f = taskManager.add { while (true) { Thread.sleep(1000) } }
 
         // Wait for the task to start
         while (!taskManager.isExecutingTask) Thread.sleep(1)
@@ -295,15 +273,17 @@
         }
       }
 
-      it("should kill the thread if interrupts failed and kill enabled") {
-        val f = taskManager.add { var x = 0; while (true) { x += 1 } }
+      // 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(true, 0)
+        taskManager.stop()
 
         // Future should return ThreadDeath when killed
         whenReady(f.failed) { result =>
diff --git a/kernel/src/main/scala/com/ibm/spark/boot/CommandLineOptions.scala b/kernel/src/main/scala/com/ibm/spark/boot/CommandLineOptions.scala
index 6b1b170..0e71a55 100644
--- a/kernel/src/main/scala/com/ibm/spark/boot/CommandLineOptions.scala
+++ b/kernel/src/main/scala/com/ibm/spark/boot/CommandLineOptions.scala
@@ -79,6 +79,11 @@
     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 options = parser.parse(args: _*)
 
   /*
@@ -130,7 +135,8 @@
           .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)
+          .flatMap(str => if (str.nonEmpty) Some(str) else None),
+        "max_interpreter_threads" -> get(_max_interpreter_threads)
     ).flatMap(removeEmptyOptions).asInstanceOf[Map[String, AnyRef]].asJava)
 
     commandLineConfig.withFallback(profileConfig).withFallback(ConfigFactory.load)
diff --git a/kernel/src/main/scala/com/ibm/spark/boot/KernelBootstrap.scala b/kernel/src/main/scala/com/ibm/spark/boot/KernelBootstrap.scala
index 6271459..f3db572 100644
--- a/kernel/src/main/scala/com/ibm/spark/boot/KernelBootstrap.scala
+++ b/kernel/src/main/scala/com/ibm/spark/boot/KernelBootstrap.scala
@@ -73,7 +73,7 @@
 
     // Initialize components needed elsewhere
     val (commStorage, commRegistrar, commManager, interpreter,
-      kernelInterpreter, kernel, sparkContext, dependencyDownloader,
+      kernel, sparkContext, dependencyDownloader,
       magicLoader, responseMap) =
       initializeComponents(
         config      = config,
@@ -81,7 +81,7 @@
         actorLoader = actorLoader
       )
     this.sparkContext = sparkContext
-    this.interpreters ++= Seq(interpreter, kernelInterpreter)
+    this.interpreters ++= Seq(interpreter)
 
     // Initialize our handlers that take care of processing messages
     initializeHandlers(
@@ -96,8 +96,7 @@
 
     // Initialize our hooks that handle various JVM events
     initializeHooks(
-      interpreter = interpreter,
-      kernelInterpreter = kernelInterpreter
+      interpreter = interpreter
     )
 
     logger.debug("Initializing security manager")
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
index 7f6c511..9011193 100644
--- a/kernel/src/main/scala/com/ibm/spark/boot/layer/ComponentInitialization.scala
+++ b/kernel/src/main/scala/com/ibm/spark/boot/layer/ComponentInitialization.scala
@@ -16,7 +16,6 @@
 
 package com.ibm.spark.boot.layer
 
-import java.io.File
 import java.util.concurrent.ConcurrentHashMap
 
 import akka.actor.ActorRef
@@ -31,15 +30,12 @@
 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.{KeyValuePairUtils, LogLike}
+import com.ibm.spark.utils.{TaskManager, KeyValuePairUtils, LogLike}
 import com.typesafe.config.Config
-import joptsimple.util.KeyValuePair
 import org.apache.spark.{SparkContext, SparkConf}
 
 import scala.collection.JavaConverters._
 
-import scala.tools.nsc.Settings
-import scala.tools.nsc.interpreter.JPrintWriter
 import scala.util.Try
 
 /**
@@ -56,7 +52,7 @@
    */
   def initializeComponents(
     config: Config, appName: String, actorLoader: ActorLoader
-  ): (CommStorage, CommRegistrar, CommManager, Interpreter, Interpreter,
+  ): (CommStorage, CommRegistrar, CommManager, Interpreter,
     Kernel, SparkContext, DependencyDownloader, MagicLoader,
     collection.mutable.Map[String, ActorRef])
 }
@@ -80,17 +76,16 @@
     val (commStorage, commRegistrar, commManager) =
       initializeCommObjects(actorLoader)
     val interpreter = initializeInterpreter(config)
-    val kernelInterpreter = initializeKernelInterpreter(config, interpreter)
     val sparkContext = initializeSparkContext(
       config, appName, actorLoader, interpreter)
     val dependencyDownloader = initializeDependencyDownloader(config)
     val magicLoader = initializeMagicLoader(
-      config, interpreter, kernelInterpreter, sparkContext, dependencyDownloader)
+      config, interpreter, sparkContext, dependencyDownloader)
     val kernel = initializeKernel(
-      actorLoader, interpreter, kernelInterpreter, commManager, magicLoader)
+      actorLoader, interpreter, commManager, magicLoader)
     val responseMap = initializeResponseMap()
-    (commStorage, commRegistrar, commManager, interpreter, kernelInterpreter,
-      kernel, sparkContext, dependencyDownloader, magicLoader, responseMap)
+    (commStorage, commRegistrar, commManager, interpreter, kernel,
+      sparkContext, dependencyDownloader, magicLoader, responseMap)
   }
 
   private def initializeCommObjects(actorLoader: ActorLoader) = {
@@ -117,13 +112,18 @@
 
   protected def initializeInterpreter(config: Config) = {
     val interpreterArgs = config.getStringList("interpreter_args").asScala.toList
+    val maxInterpreterThreads = config.getInt("max_interpreter_threads")
 
-    logger.info("Constructing interpreter with arguments: " +
-      interpreterArgs.mkString(" "))
+    logger.info(
+      s"Constructing interpreter with $maxInterpreterThreads threads and " +
+      "with arguments: " + interpreterArgs.mkString(" "))
     val interpreter = new ScalaInterpreter(interpreterArgs, Console.out)
       with StandardSparkIMainProducer
-      with StandardTaskManagerProducer
-      with StandardSettingsProducer
+      with TaskManagerProducerLike
+      with StandardSettingsProducer {
+      override def newTaskManager(): TaskManager =
+        new TaskManager(maximumWorkers = maxInterpreterThreads)
+    }
 
     logger.debug("Starting interpreter")
     interpreter.start()
@@ -131,30 +131,6 @@
     interpreter
   }
 
-  private def initializeKernelInterpreter(
-    config: Config, interpreter: Interpreter
-  ) = {
-    val interpreterArgs =
-      config.getStringList("interpreter_args").asScala.toList
-
-    // TODO: Refactor this construct to a cleaner implementation (for future
-    //       multi-interpreter design)
-    logger.info("Constructing interpreter with arguments: " +
-      interpreterArgs.mkString(" "))
-    val kernelInterpreter = new ScalaInterpreter(interpreterArgs, Console.out)
-      with StandardTaskManagerProducer
-      with StandardSettingsProducer
-      with SparkIMainProducerLike {
-      override def newSparkIMain(settings: Settings, out: JPrintWriter) = {
-        interpreter.asInstanceOf[ScalaInterpreter].sparkIMain
-      }
-    }
-    logger.debug("Starting interpreter")
-    kernelInterpreter.start()
-
-    kernelInterpreter
-  }
-
   // TODO: Think of a better way to test without exposing this
   protected[layer] def initializeSparkContext(
     config: Config, appName: String, actorLoader: ActorLoader,
@@ -283,27 +259,24 @@
 
   private def initializeKernel(
     actorLoader: ActorLoader,
-    interpreterToDoBinding: Interpreter,
-    interpreterToBind: Interpreter,
+    interpreter: Interpreter,
     commManager: CommManager,
     magicLoader: MagicLoader
   ) = {
-    //interpreter.doQuietly {
-    val kernel = new Kernel(
-      actorLoader, interpreterToBind, commManager, magicLoader
-    )
-    interpreterToDoBinding.bind(
-      "kernel", "com.ibm.spark.kernel.api.Kernel",
-      kernel, List( """@transient implicit""")
-    )
-    //}
+    val kernel = new Kernel(actorLoader, interpreter, 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, kernelInterpreter: Interpreter, sparkContext: SparkContext,
+    config: Config, interpreter: Interpreter, sparkContext: SparkContext,
     dependencyDownloader: DependencyDownloader
   ) = {
     logger.debug("Constructing magic loader")
@@ -311,8 +284,8 @@
     logger.debug("Building dependency map")
     val dependencyMap = new DependencyMap()
       .setInterpreter(interpreter)
+      .setKernelInterpreter(interpreter) // This is deprecated
       .setSparkContext(sparkContext)
-      .setKernelInterpreter(kernelInterpreter)
       .setDependencyDownloader(dependencyDownloader)
 
     logger.debug("Creating BuiltinLoader")
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
index 7e83188..2f23599 100644
--- a/kernel/src/main/scala/com/ibm/spark/boot/layer/HookInitialization.scala
+++ b/kernel/src/main/scala/com/ibm/spark/boot/layer/HookInitialization.scala
@@ -29,11 +29,8 @@
    * Initializes and registers all hooks.
    *
    * @param interpreter The main interpreter
-   * @param kernelInterpreter The interpreter bound to the kernel instance
    */
-  def initializeHooks(
-    interpreter: Interpreter, kernelInterpreter: Interpreter
-  ): Unit
+  def initializeHooks(interpreter: Interpreter): Unit
 }
 
 /**
@@ -46,18 +43,13 @@
    * Initializes and registers all hooks.
    *
    * @param interpreter The main interpreter
-   * @param kernelInterpreter The interpreter bound to the kernel instance
    */
-  def initializeHooks(
-    interpreter: Interpreter, kernelInterpreter: Interpreter
-  ): Unit = {
-    registerInterruptHook(interpreter, kernelInterpreter)
+  def initializeHooks(interpreter: Interpreter): Unit = {
+    registerInterruptHook(interpreter)
     registerShutdownHook()
   }
 
-  private def registerInterruptHook(
-    interpreter: Interpreter, kernelInterpreter: Interpreter
-  ): Unit = {
+  private def registerInterruptHook(interpreter: Interpreter): Unit = {
     val self = this
 
     import sun.misc.{Signal, SignalHandler}
@@ -73,7 +65,6 @@
         if (currentTime - lastSignalReceived > MaxSignalTime) {
           logger.info("Resetting code execution!")
           interpreter.interrupt()
-          kernelInterpreter.interrupt()
 
           // TODO: Cancel group representing current code execution
           //sparkContext.cancelJobGroup()
diff --git a/kernel/src/test/scala/com/ibm/spark/boot/CommandLineOptionsSpec.scala b/kernel/src/test/scala/com/ibm/spark/boot/CommandLineOptionsSpec.scala
index 303037b..62de035 100644
--- a/kernel/src/test/scala/com/ibm/spark/boot/CommandLineOptionsSpec.scala
+++ b/kernel/src/test/scala/com/ibm/spark/boot/CommandLineOptionsSpec.scala
@@ -27,6 +27,19 @@
 class CommandLineOptionsSpec extends FunSpec with Matchers {
 
   describe("CommandLineOptions") {
+    describe("when received --max-interpreter-threads=<int>") {
+      it("should set the configuration to the specified value") {
+        val expected = 999
+        val options = new CommandLineOptions(
+          s"--max-interpreter-threads=$expected" :: Nil
+        )
+
+        val actual = options.toConfig.getInt("max_interpreter_threads")
+
+        actual should be (expected)
+      }
+    }
+
     describe("when received --help") {
       it("should set the help flag to true") {
         val options = new CommandLineOptions("--help" :: Nil)
diff --git a/resources/compile/reference.conf b/resources/compile/reference.conf
index a677a96..376c9ed 100644
--- a/resources/compile/reference.conf
+++ b/resources/compile/reference.conf
@@ -48,3 +48,6 @@
 
 spark_configuration = ""
 spark_configuration = ${?SPARK_CONFIGURATION}
+
+max_interpreter_threads = 4
+max_interpreter_threads = ${?MAX_INTERPRETER_THREADS}
\ No newline at end of file
diff --git a/resources/test/reference.conf b/resources/test/reference.conf
index 8c47069..75ccbe8 100644
--- a/resources/test/reference.conf
+++ b/resources/test/reference.conf
@@ -49,3 +49,5 @@
 spark_configuration = ""
 spark_configuration = ${?SPARK_CONFIGURATION}
 
+max_interpreter_threads = 4
+max_interpreter_threads = ${?MAX_INTERPRETER_THREADS}