Collection of updates to SIP-14 (scala.concurrent)

Developed by Viktor Klang and Havoc Pennington

- add Promise.isCompleted

- add Future.successful and Future.failed

- add ExecutionContextExecutor and ExecutionContextExecutorService for Java interop

- remove defaultExecutionContext as default parameter value from promise and future

- add ExecutionContext.Implicits.global which must be explicitly imported, rather
  than the previous always-available value for the implicit EC

- remove currentExecutionContext, since it could create bugs by being
  out of sync with the implicit ExecutionContext

- remove Future task batching (_taskStack) and Future.releaseStack

    This optimization should instead be implemented either in
    a specific thread pool or in a specific ExecutionContext.
    Some pools or ExecutionContexts may not want or need it.
    In this patch, the defaultExecutionContext does not
    keep the batching optimization. Whether it should
    have it should perhaps be determined through benchmarking.

- move internalBlockingCall to BlockContext and remove currentExecutionContext

    In this patch, BlockContext must be implemented by Thread.currentThread,
    so the thread pool is the only place you can add custom hooks
    to be run when blocking.
    We implement BlockContext for the default ForkJoinWorkerThread in terms of
    ForkJoinPool.ManagedBlocker.

- add public BlockContext.current and BlockContext.withBlockContext

    These allow an ExecutionContext or other code to override
    the BlockContext for the current thread. With this
    API, the BlockContext is customizable without
    creating a new pool of threads.

    BlockContext.current is needed to obtain the previous
    BlockContext before you push, so you can "chain up" to
    it if desired.

    BlockContext.withBlockContext is used to override the context
    for a given piece of code.

- move isFutureThrowable into impl.Future

- add implicitNotFound to ExecutionContext

- remove default global EC from future {} and promise {}

- add ExecutionContext.global for explicit use of the global default EC,
  replaces defaultExecutionContext

- add a timeout to scala-concurrent-tck tests that block on SyncVar
  (so tests time out rather than hang)

- insert blocking{} calls into concurrent tck to fix deadlocking

- add NonFatal.apply and tests for NonFatal

- add OnCompleteRunnable marker trait

    This would allow an ExecutionContext to distinguish a Runnable originating
    from Future.onComplete (all callbacks on Future end up going through
    onComplete).

- rename ListenerRunnable to CallbackRunnable and use for KeptPromise too
  Just adds some clarity and consistency.

3 files changed, 59 insertions, 138 deletions

diff --git a/src/library/scala/concurrent/impl/ExecutionContextImpl.scala b/src/library/scala/concurrent/impl/ExecutionContextImpl.scala
index 4c6347dce0..551a444425 100644
--- a/src/library/scala/concurrent/impl/ExecutionContextImpl.scala
+++ b/src/library/scala/concurrent/impl/ExecutionContextImpl.scala
@@ -13,34 +13,34 @@ package scala.concurrent.impl
 import java.util.concurrent.{ Callable, Executor, ExecutorService, Executors, ThreadFactory, TimeUnit }
 import java.util.Collection
 import scala.concurrent.forkjoin._
-import scala.concurrent.{ ExecutionContext, Awaitable }
+import scala.concurrent.{ BlockContext, ExecutionContext, Awaitable, ExecutionContextExecutor, ExecutionContextExecutorService }
 import scala.concurrent.util.Duration
 import scala.util.control.NonFatal
 
 
 
-private[scala] class ExecutionContextImpl private[impl] (es: Executor, reporter: Throwable => Unit) extends ExecutionContext with Executor {
+private[scala] class ExecutionContextImpl private[impl] (es: Executor, reporter: Throwable => Unit) extends ExecutionContextExecutor {
 
   val executor: Executor = es match {
     case null => createExecutorService
     case some => some
   }
-  
-  // to ensure that the current execution context thread local is properly set
-  def executorsThreadFactory = new ThreadFactory {
-    def newThread(r: Runnable) = new Thread(new Runnable {
-      override def run() {
-        scala.concurrent.currentExecutionContext.set(ExecutionContextImpl.this)
-        r.run()
-      }
-    })
-  }
-  
-  // to ensure that the current execution context thread local is properly set
+
+  // Implement BlockContext on FJP threads
   def forkJoinPoolThreadFactory = new ForkJoinPool.ForkJoinWorkerThreadFactory {
-    def newThread(fjp: ForkJoinPool) = new ForkJoinWorkerThread(fjp) {
-      override def onStart() {
-        scala.concurrent.currentExecutionContext.set(ExecutionContextImpl.this)
+    def newThread(fjp: ForkJoinPool) = new ForkJoinWorkerThread(fjp) with BlockContext {
+      override def internalBlockingCall[T](awaitable: Awaitable[T], atMost: Duration): T = {
+        var result: T = null.asInstanceOf[T]
+        ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker {
+          @volatile var isdone = false
+          def block(): Boolean = {
+            result = awaitable.result(atMost)(scala.concurrent.Await.canAwaitEvidence) // FIXME what happens if there's an exception thrown here?
+            isdone = true
+            true
+          }
+          def isReleasable = isdone
+        })
+        result
       }
     }
   }
@@ -68,7 +68,7 @@ private[scala] class ExecutionContextImpl private[impl] (es: Executor, reporter:
     case NonFatal(t) =>
       System.err.println("Failed to create ForkJoinPool for the default ExecutionContext, falling back to Executors.newCachedThreadPool")
       t.printStackTrace(System.err)
-      Executors.newCachedThreadPool(executorsThreadFactory) //FIXME use the same desired parallelism here too?
+      Executors.newCachedThreadPool() //FIXME use the same desired parallelism here too?
   }
 
   def execute(runnable: Runnable): Unit = executor match {
@@ -84,27 +84,6 @@ private[scala] class ExecutionContextImpl private[impl] (es: Executor, reporter:
     case generic => generic execute runnable
   }
 
-  def internalBlockingCall[T](awaitable: Awaitable[T], atMost: Duration): T = {
-    Future.releaseStack(this)
-    
-    executor match {
-      case fj: ForkJoinPool =>
-        var result: T = null.asInstanceOf[T]
-        ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker { 
-          @volatile var isdone = false
-          def block(): Boolean = {
-            result = awaitable.result(atMost)(scala.concurrent.Await.canAwaitEvidence) // FIXME what happens if there's an exception thrown here?
-            isdone = true
-            true
-          }
-          def isReleasable = isdone
-        })
-        result
-      case _ =>
-        awaitable.result(atMost)(scala.concurrent.Await.canAwaitEvidence)
-    }
-  }
-
   def reportFailure(t: Throwable) = reporter(t)
 }
 
@@ -112,8 +91,8 @@ private[scala] class ExecutionContextImpl private[impl] (es: Executor, reporter:
 private[concurrent] object ExecutionContextImpl {
 
   def fromExecutor(e: Executor, reporter: Throwable => Unit = ExecutionContext.defaultReporter): ExecutionContextImpl = new ExecutionContextImpl(e, reporter)
-  def fromExecutorService(es: ExecutorService, reporter: Throwable => Unit = ExecutionContext.defaultReporter): ExecutionContextImpl with ExecutorService =
-    new ExecutionContextImpl(es, reporter) with ExecutorService {
+  def fromExecutorService(es: ExecutorService, reporter: Throwable => Unit = ExecutionContext.defaultReporter): ExecutionContextImpl with ExecutionContextExecutorService =
+    new ExecutionContextImpl(es, reporter) with ExecutionContextExecutorService {
       final def asExecutorService: ExecutorService = executor.asInstanceOf[ExecutorService]
       override def execute(command: Runnable) = executor.execute(command)
       override def shutdown() { asExecutorService.shutdown() }
diff --git a/src/library/scala/concurrent/impl/Future.scala b/src/library/scala/concurrent/impl/Future.scala
index 8012ea6a93..073e6c4c9f 100644
--- a/src/library/scala/concurrent/impl/Future.scala
+++ b/src/library/scala/concurrent/impl/Future.scala
@@ -46,11 +46,19 @@ private[concurrent] object Future {
   
   def boxedType(c: Class[_]): Class[_] = if (c.isPrimitive) toBoxed(c) else c
 
-  private[impl] class PromiseCompletingTask[T](override val executor: ExecutionContext, body: => T)
-    extends Task {
+  // TODO rename appropriately and make public
+  private[concurrent] def isFutureThrowable(t: Throwable) = t match {
+    case e: Error                               => false
+    case t: scala.util.control.ControlThrowable => false
+    case i: InterruptedException                => false
+    case _                                      => true
+  }
+
+  private[impl] class PromiseCompletingRunnable[T](body: => T)
+    extends Runnable {
     val promise = new Promise.DefaultPromise[T]()
 
-    protected override def task() = {
+    override def run() = {
       promise complete {
         try Right(body) catch {
           case NonFatal(e) =>
@@ -63,90 +71,8 @@ private[concurrent] object Future {
   }
 
   def apply[T](body: =>T)(implicit executor: ExecutionContext): Future[T] = {
-    val task = new PromiseCompletingTask(executor, body)
-    task.dispatch()
-
-    task.promise.future
-  }
-
-  private[impl] val throwableId: Throwable => Throwable = identity _
-
-  // an optimization for batching futures
-  // TODO we should replace this with a public queue,
-  // so that it can be stolen from
-  // OR: a push to the local task queue should be so cheap that this is
-  // not even needed, but stealing is still possible
-
-  private[impl] case class TaskStack(stack: Stack[Task], executor: ExecutionContext)
-
-  private val _taskStack = new ThreadLocal[TaskStack]()
-
-  private[impl] trait Task extends Runnable {
-    def executor: ExecutionContext
-
-    // run the original callback (no dispatch)
-    protected def task(): Unit
-
-    // we implement Runnable to avoid creating
-    // an extra object. run() runs ourselves with
-    // a TaskStack pushed, and then runs any
-    // other tasks that show up in the stack.
-    final override def run() = {
-      try {
-        val taskStack = TaskStack(Stack[Task](this), executor)
-        _taskStack set taskStack
-        while (taskStack.stack.nonEmpty) {
-          val next = taskStack.stack.pop()
-          require(next.executor eq executor)
-          try next.task() catch { case NonFatal(e) => executor reportFailure e }
-        }
-      } finally {
-        _taskStack.remove()
-      }
-    }
-
-    // send the task to the running executor.execute() via
-    // _taskStack, or start a new executor.execute()
-    def dispatch(force: Boolean = false): Unit =
-      _taskStack.get match {
-        case stack if (stack ne null) && (executor eq stack.executor) && !force => stack.stack push this
-        case _ => executor.execute(this)
-      }
-  }
-
-  private[impl] class ReleaseTask(override val executor: ExecutionContext, val elems: List[Task])
-    extends Task {
-    protected override def task() = {
-      _taskStack.get.stack.elems = elems
-    }
-  }
-
-  private[impl] def releaseStack(executor: ExecutionContext): Unit =
-    _taskStack.get match {
-      case stack if (stack ne null) && stack.stack.nonEmpty =>
-        val tasks = stack.stack.elems
-        stack.stack.clear()
-        _taskStack.remove()
-        val release = new ReleaseTask(executor, tasks)
-        release.dispatch(force=true)
-      case null =>
-        // do nothing - there is no local batching stack anymore
-      case _ =>
-        _taskStack.remove()
-    }
-
-  private[impl] class OnCompleteTask[T](override val executor: ExecutionContext, val onComplete: (Either[Throwable, T]) => Any)
-    extends Task {
-    private var value: Either[Throwable, T] = null
-
-    protected override def task() = {
-      require(value ne null) // dispatch(value) must be called before dispatch()
-      onComplete(value)
-    }
-
-    def dispatch(value: Either[Throwable, T]): Unit = {
-      this.value = value
-      dispatch()
-    }
+    val runnable = new PromiseCompletingRunnable(body)
+    executor.execute(runnable)
+    runnable.promise.future
   }
 }
diff --git a/src/library/scala/concurrent/impl/Promise.scala b/src/library/scala/concurrent/impl/Promise.scala
index c5060a2368..3ac34bef8a 100644
--- a/src/library/scala/concurrent/impl/Promise.scala
+++ b/src/library/scala/concurrent/impl/Promise.scala
@@ -11,11 +11,12 @@ package scala.concurrent.impl
 
 
 import java.util.concurrent.TimeUnit.{ NANOSECONDS, MILLISECONDS }
-import scala.concurrent.{ Awaitable, ExecutionContext, blocking, CanAwait, TimeoutException, ExecutionException }
+import scala.concurrent.{ Awaitable, ExecutionContext, blocking, CanAwait, OnCompleteRunnable, TimeoutException, ExecutionException }
 //import scala.util.continuations._
 import scala.concurrent.util.Duration
 import scala.util
 import scala.annotation.tailrec
+import scala.util.control.NonFatal
 //import scala.concurrent.NonDeterministic
 
 
@@ -24,6 +25,21 @@ private[concurrent] trait Promise[T] extends scala.concurrent.Promise[T] with Fu
   def future: this.type = this
 }
 
+private class CallbackRunnable[T](val executor: ExecutionContext, val onComplete: (Either[Throwable, T]) => Any) extends Runnable with OnCompleteRunnable {
+  // must be filled in before running it
+  var value: Either[Throwable, T] = null
+
+  override def run() = {
+    require(value ne null) // must set value to non-null before running!
+    try onComplete(value) catch { case NonFatal(e) => executor reportFailure e }
+  }
+
+  def executeWithValue(v: Either[Throwable, T]): Unit = {
+    require(value eq null) // can't complete it twice
+    value = v
+    executor.execute(this)
+  }
+}
 
 object Promise {
 
@@ -94,10 +110,10 @@ object Promise {
       val resolved = resolveEither(value)
       (try {
         @tailrec
-        def tryComplete(v: Either[Throwable, T]): List[Future.OnCompleteTask[T]] = {
+        def tryComplete(v: Either[Throwable, T]): List[CallbackRunnable[T]] = {
           getState match {
             case raw: List[_] =>
-              val cur = raw.asInstanceOf[List[Future.OnCompleteTask[T]]]
+              val cur = raw.asInstanceOf[List[CallbackRunnable[T]]]
               if (updateState(cur, v)) cur else tryComplete(v)
             case _ => null
           }
@@ -107,19 +123,19 @@ object Promise {
         synchronized { notifyAll() } //Notify any evil blockers
       }) match {
         case null             => false
-        case cs if cs.isEmpty => true
-        case cs               => cs.foreach(c => c.dispatch(resolved)); true
+        case rs if rs.isEmpty => true
+        case rs               => rs.foreach(r => r.executeWithValue(resolved)); true
       }
     }
 
     def onComplete[U](func: Either[Throwable, T] => U)(implicit executor: ExecutionContext): Unit = {
-      val bound = new Future.OnCompleteTask[T](executor, func)
+      val runnable = new CallbackRunnable[T](executor, func)
 
       @tailrec //Tries to add the callback, if already completed, it dispatches the callback to be executed
       def dispatchOrAddCallback(): Unit =
         getState match {
-          case r: Either[_, _]    => bound.dispatch(r.asInstanceOf[Either[Throwable, T]])
-          case listeners: List[_] => if (updateState(listeners, bound :: listeners)) () else dispatchOrAddCallback()
+          case r: Either[_, _]    => runnable.executeWithValue(r.asInstanceOf[Either[Throwable, T]])
+          case listeners: List[_] => if (updateState(listeners, runnable :: listeners)) () else dispatchOrAddCallback()
         }
       dispatchOrAddCallback()
     }
@@ -139,7 +155,7 @@ object Promise {
 
     def onComplete[U](func: Either[Throwable, T] => U)(implicit executor: ExecutionContext): Unit = {
       val completedAs = value.get
-      (new Future.OnCompleteTask(executor, func)).dispatch(completedAs)
+      (new CallbackRunnable(executor, func)).executeWithValue(completedAs)
     }
 
     def ready(atMost: Duration)(implicit permit: CanAwait): this.type = this