Merge pull request scala#856 from havocp/sip14-execution-changes

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

Author: Adriaan Moors

src/library/scala/collection/parallel/TaskSupport.scala

@@ -48,7 +48,7 @@ extends TaskSupport with AdaptiveWorkStealingThreadPoolTasks
  *  By default, parallel collections are parametrized with this task support object, so parallel collections
  *  share the same execution context backend as the rest of the `scala.concurrent` package.
  */
-class ExecutionContextTaskSupport(val environment: ExecutionContext = scala.concurrent.defaultExecutionContext)
+class ExecutionContextTaskSupport(val environment: ExecutionContext = scala.concurrent.ExecutionContext.global)
 extends TaskSupport with ExecutionContextTasks
 
 

src/library/scala/concurrent/BlockContext.scala

@@ -0,0 +1,81 @@
+/*                     __                                               *\
+**     ________ ___   / /  ___     Scala API                            **
+**    / __/ __// _ | / /  / _ |    (c) 2003-2012, LAMP/EPFL             **
+**  __\ \/ /__/ __ |/ /__/ __ |    http://scala-lang.org/               **
+** /____/\___/_/ |_/____/_/ | |                                         **
+**                          |/                                          **
+\*                                                                      */
+
+package scala.concurrent
+
+import java.lang.Thread
+import scala.concurrent.util.Duration
+
+/**
+ * A context to be notified by `scala.concurrent.blocking()` when
+ * a thread is about to block. In effect this trait provides
+ * the implementation for `scala.concurrent.blocking()`. `scala.concurrent.blocking()`
+ * locates an instance of `BlockContext` by first looking for one
+ * provided through `BlockContext.withBlockContext()` and failing that,
+ * checking whether `Thread.currentThread` is an instance of `BlockContext`.
+ * So a thread pool can have its `java.lang.Thread` instances implement
+ * `BlockContext`. There's a default `BlockContext` used if the thread
+ * doesn't implement `BlockContext`.
+ *
+ * Typically, you'll want to chain to the previous `BlockContext`,
+ * like this:
+ * {{{
+ *  val oldContext = BlockContext.current
+ *  val myContext = new BlockContext {
+ *    override def internalBlockingCall[T](awaitable: Awaitable[T], atMost: Duration): T = {
+ *      // you'd have code here doing whatever you need to do
+ *      // when the thread is about to block.
+ *      // Then you'd chain to the previous context:
+ *      oldContext.internalBlockingCall(awaitable, atMost)
+ *    }
+ *  }
+ *  BlockContext.withBlockContext(myContext) {
+ *    // then this block runs with myContext as the handler
+ *    // for scala.concurrent.blocking
+ *  }
+ *  }}}
+ */
+trait BlockContext {
+
+  /** Used internally by the framework; blocks execution for at most
+   * `atMost` time while waiting for an `awaitable` object to become ready.
+   *
+   * Clients should use `scala.concurrent.blocking` instead; this is
+   * the implementation of `scala.concurrent.blocking`, generally
+   * provided by a `scala.concurrent.ExecutionContext` or `java.util.concurrent.Executor`.
+   */
+  def internalBlockingCall[T](awaitable: Awaitable[T], atMost: Duration): T
+}
+
+object BlockContext {
+  private object DefaultBlockContext extends BlockContext {
+    override def internalBlockingCall[T](awaitable: Awaitable[T], atMost: Duration): T =
+      awaitable.result(atMost)(Await.canAwaitEvidence)
+  }
+
+  private val contextLocal = new ThreadLocal[BlockContext]() {
+    override def initialValue = Thread.currentThread match {
+      case ctx: BlockContext => ctx
+      case _ => DefaultBlockContext
+    }
+  }
+
+  /** Obtain the current thread's current `BlockContext`. */
+  def current: BlockContext = contextLocal.get
+
+  /** Pushes a current `BlockContext` while executing `body`. */
+  def withBlockContext[T](blockContext: BlockContext)(body: => T): T = {
+    val old = contextLocal.get
+    try {
+      contextLocal.set(blockContext)
+      body
+    } finally {
+      contextLocal.set(old)
+    }
+  }
+}

src/library/scala/concurrent/ConcurrentPackageObject.scala

@@ -17,23 +17,6 @@ import language.implicitConversions
 /** This package object contains primitives for concurrent and parallel programming.
  */
 abstract class ConcurrentPackageObject {
-  /** A global execution environment for executing lightweight tasks.
-   */
-  lazy val defaultExecutionContext: ExecutionContext with Executor = impl.ExecutionContextImpl.fromExecutor(null: Executor)
-
-  val currentExecutionContext = new ThreadLocal[ExecutionContext]
-  
-  val handledFutureException: PartialFunction[Throwable, Throwable] = {
-    case t: Throwable if isFutureThrowable(t) => t
-  }
-
-  // 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
-  }
 
   /* concurrency constructs */
 
@@ -46,17 +29,15 @@ abstract class ConcurrentPackageObject {
    *  @param execctx  the execution context on which the future is run
    *  @return         the `Future` holding the result of the computation
    */
-  def future[T](body: =>T)(implicit execctx: ExecutionContext = defaultExecutionContext): Future[T] =
-    Future[T](body)
+  def future[T](body: =>T)(implicit execctx: ExecutionContext): Future[T] = Future[T](body)
 
   /** Creates a promise object which can be completed with a value.
    *  
    *  @tparam T       the type of the value in the promise
    *  @param execctx  the execution context on which the promise is created on
    *  @return         the newly created `Promise` object
    */
-  def promise[T]()(implicit execctx: ExecutionContext = defaultExecutionContext): Promise[T] =
-    Promise[T]()
+  def promise[T]()(implicit execctx: ExecutionContext): Promise[T] = Promise[T]()
 
   /** Used to block on a piece of code which potentially blocks.
    *
@@ -67,8 +48,7 @@ abstract class ConcurrentPackageObject {
    *  - InterruptedException - in the case that a wait within the blockable object was interrupted
    *  - TimeoutException - in the case that the blockable object timed out
    */
-  def blocking[T](body: =>T): T =
-    blocking(impl.Future.body2awaitable(body), Duration.Inf)
+  def blocking[T](body: =>T): T = blocking(impl.Future.body2awaitable(body), Duration.Inf)
 
   /** Blocks on an awaitable object.
    *
@@ -79,12 +59,8 @@ abstract class ConcurrentPackageObject {
    *  - InterruptedException - in the case that a wait within the blockable object was interrupted
    *  - TimeoutException - in the case that the blockable object timed out
    */
-  def blocking[T](awaitable: Awaitable[T], atMost: Duration): T = {
-    currentExecutionContext.get match {
-      case null => awaitable.result(atMost)(Await.canAwaitEvidence)
-      case ec => ec.internalBlockingCall(awaitable, atMost)
-    }
-  }
+  def blocking[T](awaitable: Awaitable[T], atMost: Duration): T =
+    BlockContext.current.internalBlockingCall(awaitable, atMost)
 
   @inline implicit final def int2durationops(x: Int): DurationOps = new DurationOps(x)
 }

src/library/scala/concurrent/DelayedLazyVal.scala

@@ -23,7 +23,7 @@ package scala.concurrent
  *  @author  Paul Phillips
  *  @version 2.8
  */
-class DelayedLazyVal[T](f: () => T, body: => Unit) {
+class DelayedLazyVal[T](f: () => T, body: => Unit){
   @volatile private[this] var _isDone = false
   private[this] lazy val complete = f()
 
@@ -39,7 +39,8 @@ class DelayedLazyVal[T](f: () => T, body: => Unit) {
    */
   def apply(): T = if (isDone) complete else f()
 
-  // TODO replace with scala.concurrent.future { ... }
+  // FIXME need to take ExecutionContext in constructor
+  import ExecutionContext.Implicits.global
   future {
     body
     _isDone = true

src/library/scala/concurrent/ExecutionContext.scala

@@ -9,58 +9,80 @@
 package scala.concurrent
 
 
-
-import java.util.concurrent.atomic.{ AtomicInteger }
-import java.util.concurrent.{ Executors, Future => JFuture, Callable, ExecutorService, Executor }
+import java.util.concurrent.{ ExecutorService, Executor }
 import scala.concurrent.util.Duration
-import scala.concurrent.forkjoin.{ ForkJoinPool, RecursiveTask => FJTask, RecursiveAction, ForkJoinWorkerThread }
-import scala.collection.generic.CanBuildFrom
-import collection._
-
-
+import scala.annotation.implicitNotFound
 
+/**
+ * An `ExecutionContext` is an abstraction over an entity that can execute program logic.
+ */
+@implicitNotFound("Cannot find an implicit ExecutionContext, either require one yourself or import ExecutionContext.Implicits.global")
 trait ExecutionContext {
 
   /** Runs a block of code on this execution context.
    */
   def execute(runnable: Runnable): Unit
 
-  /** Used internally by the framework - blocks execution for at most `atMost` time while waiting
-   *  for an `awaitable` object to become ready.
-   *  
-   *  Clients should use `scala.concurrent.blocking` instead.
-   */
-  def internalBlockingCall[T](awaitable: Awaitable[T], atMost: Duration): T
-  
   /** Reports that an asynchronous computation failed.
    */
   def reportFailure(t: Throwable): Unit
 
 }
 
+/**
+ * Union interface since Java does not support union types
+ */
+trait ExecutionContextExecutor extends ExecutionContext with Executor
+
+/**
+ * Union interface since Java does not support union types
+ */
+trait ExecutionContextExecutorService extends ExecutionContextExecutor with ExecutorService
+
 
 /** Contains factory methods for creating execution contexts.
  */
 object ExecutionContext {
-  
-  implicit def defaultExecutionContext: ExecutionContext = scala.concurrent.defaultExecutionContext
-  
+  /**
+   * The `ExecutionContext` associated with the current `Thread`
+   */
+  val currentExecutionContext: ThreadLocal[ExecutionContext] = new ThreadLocal //FIXME might want to set the initial value to an executionContext that throws an exception on execute and warns that it's not set
+
+  /**
+   * This is the explicit global ExecutionContext,
+   * call this when you want to provide the global ExecutionContext explicitly
+   */
+  def global: ExecutionContextExecutor = Implicits.global
+
+  object Implicits {
+    /**
+     * This is the implicit global ExecutionContext,
+     * import this when you want to provide the global ExecutionContext implicitly
+     */
+    implicit lazy val global: ExecutionContextExecutor = impl.ExecutionContextImpl.fromExecutor(null: Executor)
+  }
+    
   /** Creates an `ExecutionContext` from the given `ExecutorService`.
    */
-  def fromExecutorService(e: ExecutorService, reporter: Throwable => Unit = defaultReporter): ExecutionContext with ExecutorService =
+  def fromExecutorService(e: ExecutorService, reporter: Throwable => Unit): ExecutionContextExecutorService =
     impl.ExecutionContextImpl.fromExecutorService(e, reporter)
+
+  /** Creates an `ExecutionContext` from the given `ExecutorService` with the default Reporter.
+   */
+  def fromExecutorService(e: ExecutorService): ExecutionContextExecutorService = fromExecutorService(e, defaultReporter)
 
   /** Creates an `ExecutionContext` from the given `Executor`.
    */
-  def fromExecutor(e: Executor, reporter: Throwable => Unit = defaultReporter): ExecutionContext with Executor =
+  def fromExecutor(e: Executor, reporter: Throwable => Unit): ExecutionContextExecutor =
     impl.ExecutionContextImpl.fromExecutor(e, reporter)
+
+  /** Creates an `ExecutionContext` from the given `Executor` with the default Reporter.
+   */
+  def fromExecutor(e: Executor): ExecutionContextExecutor = fromExecutor(e, defaultReporter)
 
-  def defaultReporter: Throwable => Unit = {
-    // re-throwing `Error`s here causes an exception handling test to fail.
-    //case e: Error => throw e
-    case t => t.printStackTrace()
-  }
-  
+  /** The default reporter simply prints the stack trace of the `Throwable` to System.err.
+   */
+  def defaultReporter: Throwable => Unit = { case t => t.printStackTrace() }
 }
 
 

src/library/scala/concurrent/Future.scala

@@ -8,7 +8,7 @@
 
 package scala.concurrent
 
-
+import language.higherKinds
 
 import java.util.concurrent.{ ConcurrentLinkedQueue, TimeUnit, Callable }
 import java.util.concurrent.TimeUnit.{ NANOSECONDS => NANOS, MILLISECONDS ⇒ MILLIS }
@@ -23,11 +23,9 @@ import scala.Option
 import scala.util.{Try, Success, Failure}
 
 import scala.annotation.tailrec
-import scala.collection.mutable.Stack
 import scala.collection.mutable.Builder
 import scala.collection.generic.CanBuildFrom
 import scala.reflect.ClassTag
-import language.higherKinds
 
 
 
@@ -138,7 +136,7 @@ trait Future[+T] extends Awaitable[T] {
    *  $callbackInContext
    */
   def onFailure[U](callback: PartialFunction[Throwable, U])(implicit executor: ExecutionContext): Unit = onComplete {
-    case Left(t) if (isFutureThrowable(t) && callback.isDefinedAt(t)) => callback(t)
+    case Left(t) if (impl.Future.isFutureThrowable(t) && callback.isDefinedAt(t)) => callback(t)
     case _ =>
   }(executor)
 
@@ -580,6 +578,20 @@ object Future {
     classOf[Double]  -> classOf[jl.Double],
     classOf[Unit]    -> classOf[scala.runtime.BoxedUnit]
   )
+
+  /** Creates an already completed Future with the specified exception.
+   *  
+   *  @tparam T       the type of the value in the future
+   *  @return         the newly created `Future` object
+   */
+  def failed[T](exception: Throwable): Future[T] = Promise.failed(exception).future
+
+  /** Creates an already completed Future with the specified result.
+   *  
+   *  @tparam T       the type of the value in the future
+   *  @return         the newly created `Future` object
+   */
+  def successful[T](result: T): Future[T] = Promise.successful(result).future
 
   /** Starts an asynchronous computation and returns a `Future` object with the result of that computation.
   *
@@ -710,5 +722,12 @@ object Future {
   }
 }
 
-
+/** A marker indicating that a `java.lang.Runnable` provided to `scala.concurrent.ExecutionContext`
+ * wraps a callback provided to `Future.onComplete`.
+ * All callbacks provided to a `Future` end up going through `onComplete`, so this allows an
+ * `ExecutionContext` to special-case callbacks that were executed by `Future` if desired.
+ */
+trait OnCompleteRunnable {
+  self: Runnable =>
+}
 

src/library/scala/concurrent/Promise.scala

@@ -34,6 +34,15 @@ trait Promise[T] {
    */
   def future: Future[T]
 
+  /** Returns whether the promise has already been completed with
+   *  a value or an exception.
+   *
+   *  $nonDeterministic
+   *
+   *  @return    `true` if the promise is already completed, `false` otherwise
+   */
+  def isCompleted: Boolean
+
   /** Completes the promise with either an exception or a value.
    *
    *  @param result     Either the value or the exception to complete the promise with.