This update / revision of the Future and Promise implementation:

Introduces,
 Future.delegate[A](=> Future[A])(implicit ec: ExecutionContext): Future[A]
  - this allows for construction of Futures which is deferred
    onto the supplied ExecutionContext
  - Equivalent to: Future.unit.flatMap(_ => expr)

Makes Futures correctly handle RejectedExecutionExceptions,
from ExecutionContext.execute.

Makes Futures correctly handle Thread.interrupt()'s.

Needs more external verification on how much it improves performance
for most, if not all, combinators.

Introduces a different implementation of Linking, with dedicated link
nodes—which are GC:ed upon completion.

Introduces a different implementation of the transformations, instead of
relying primarily on transformWith or onComplete, the DefaultPromise
specializes practically all combinators. It also uses a manually encoded,
unrolled, dispatch instead of polymorphism, and uses an O(n) linking
concat (reverse-order) to avoid creating arbitrary-depth callback tries.

Duration.Undefined doesn't equal Duration.Undefined,
so switching to eq (internal change).

Author: viktorklang

library/src/scala/concurrent/BatchingExecutor.scala

@@ -8,8 +8,17 @@
 
 package scala.concurrent
 
+import java.util.ArrayDeque
 import java.util.concurrent.Executor
 import scala.annotation.tailrec
+import scala.util.control.NonFatal
+
+/**
+ * Marker trait to indicate that a Runnable is Batchable by BatchingExecutors
+ */
+trait Batchable {
+  self: Runnable =>
+}
 
 /**
  * Mixin trait for an Executor
@@ -39,79 +48,90 @@ import scala.annotation.tailrec
  * WARNING: The underlying Executor's execute-method must not execute the submitted Runnable
  * in the calling thread synchronously. It must enqueue/handoff the Runnable.
  */
-private[concurrent] trait BatchingExecutor extends Executor {
+ private[concurrent] trait BatchingExecutor extends Executor {
+  private[this] final val _tasksLocal = new ThreadLocal[Batch]()
+
+  private[this] final class Batch(capacity: Int) extends ArrayDeque[Runnable](capacity) with Runnable with BlockContext with (BlockContext => Unit) {
+    private[this] final var parentBlockContext: BlockContext = _
 
-  // invariant: if "_tasksLocal.get ne null" then we are inside BatchingRunnable.run; if it is null, we are outside
-  private[this] val _tasksLocal = new ThreadLocal[List[Runnable]]()
+    def this(r: Runnable, capacity: Int) = {
+      this(capacity)
+      addLast(r)
+    }
+
+    final def executor: BatchingExecutor = BatchingExecutor.this
 
-  private class Batch(val initial: List[Runnable]) extends Runnable with BlockContext {
-    private[this] var parentBlockContext: BlockContext = _
     // this method runs in the delegate ExecutionContext's thread
-    override def run(): Unit = {
-      require(_tasksLocal.get eq null)
-
-      val prevBlockContext = BlockContext.current
-      BlockContext.withBlockContext(this) {
-        try {
-          parentBlockContext = prevBlockContext
-
-          @tailrec def processBatch(batch: List[Runnable]): Unit = batch match {
-            case Nil => ()
-            case head :: tail =>
-              _tasksLocal set tail
-              try {
-                head.run()
-              } catch {
-                case t: Throwable =>
-                  // if one task throws, move the
-                  // remaining tasks to another thread
-                  // so we can throw the exception
-                  // up to the invoking executor
-                  val remaining = _tasksLocal.get
-                  _tasksLocal set Nil
-                  unbatchedExecute(new Batch(remaining)) //TODO what if this submission fails?
-                  throw t // rethrow
-              }
-              processBatch(_tasksLocal.get) // since head.run() can add entries, always do _tasksLocal.get here
-          }
-
-          processBatch(initial)
-        } finally {
-          _tasksLocal.remove()
-          parentBlockContext = null
-        }
+    override final def run(): Unit = BlockContext.usingBlockContext(this)(this)
+
+    override final def apply(prevBlockContext: BlockContext): Unit = {
+      //This invariant needs to hold: require(_tasksLocal.get eq null)
+      parentBlockContext = prevBlockContext
+      try {
+        _tasksLocal.set(this)
+        runAll()
+        _tasksLocal.remove() // Will be cleared in the throwing-case by runAll()
+      } finally {
+        parentBlockContext = null
       }
     }
 
+    @tailrec private[this] final def runAll(): Unit = {
+      val next = pollLast()
+      if (next ne null) {
+        try next.run() catch {
+          case t: Throwable =>
+            parentBlockContext = null // Need to reset this before re-submitting it
+            _tasksLocal.remove() // If unbatchedExecute runs synchronously
+            handleRunFailure(t)
+         }
+        runAll()
+      }
+    }
+
+    private[this] final def handleRunFailure(cause: Throwable): Nothing =
+      if (NonFatal(cause) || cause.isInstanceOf[InterruptedException]) {
+        try unbatchedExecute(this) catch {
+          case inner: Throwable =>
+            if (NonFatal(inner)) {
+              val e = new ExecutionException("Non-fatal error occurred and resubmission failed, see suppressed exception.", cause)
+              e.addSuppressed(inner)
+              throw e
+            } else throw inner
+        }
+        throw cause
+      } else throw cause
+
     override def blockOn[T](thunk: => T)(implicit permission: CanAwait): T = {
-      // if we know there will be blocking, we don't want to keep tasks queued up because it could deadlock.
-      {
-        val tasks = _tasksLocal.get
-        _tasksLocal set Nil
-        if ((tasks ne null) && tasks.nonEmpty)
-          unbatchedExecute(new Batch(tasks))
+      val pbc = parentBlockContext
+      if(!isEmpty) { // if we know there will be blocking, we don't want to keep tasks queued up because it could deadlock.
+        val b = new Batch(math.max(4, this.size))
+        b.addAll(this)
+        this.clear()
+        unbatchedExecute(b)
       }
 
-      // now delegate the blocking to the previous BC
-      require(parentBlockContext ne null)
-      parentBlockContext.blockOn(thunk)
+      if (pbc ne null) pbc.blockOn(thunk) // now delegate the blocking to the previous BC
+      else {
+        try thunk finally throw new IllegalStateException("BUG in BatchingExecutor.Batch: parentBlockContext is null")
+      }
     }
   }
 
   protected def unbatchedExecute(r: Runnable): Unit
 
-  override def execute(runnable: Runnable): Unit = {
-    if (batchable(runnable)) { // If we can batch the runnable
-      _tasksLocal.get match {
-        case null => unbatchedExecute(new Batch(runnable :: Nil)) // If we aren't in batching mode yet, enqueue batch
-        case some => _tasksLocal.set(runnable :: some) // If we are already in batching mode, add to batch
-      }
-    } else unbatchedExecute(runnable) // If not batchable, just delegate to underlying
+  private[this] final def batchedExecute(runnable: Runnable): Unit = {
+    val b = _tasksLocal.get
+    if (b ne null) b.addLast(runnable)
+    else unbatchedExecute(new Batch(runnable, 4))
   }
 
-  /** Override this to define which runnables will be batched. */
-  def batchable(runnable: Runnable): Boolean = runnable match {
-    case _: OnCompleteRunnable => true
-    case _                     => false
-  }
-}
+  override def execute(runnable: Runnable): Unit =
+    if(batchable(runnable)) batchedExecute(runnable)
+    else unbatchedExecute(runnable)
+
+  /** Override this to define which runnables will be batched.
+    * By default it tests the Runnable for being an instance of [Batchable].
+   **/
+  protected def batchable(runnable: Runnable): Boolean = runnable.isInstanceOf[Batchable]
+}

library/src/scala/concurrent/BlockContext.scala

@@ -41,40 +41,48 @@ package scala.concurrent
 trait BlockContext {
 
   /** Used internally by the framework;
-   * Designates (and eventually executes) a thunk which potentially blocks the calling `java.lang.Thread`.
-   *
-   * Clients must use `scala.concurrent.blocking` or `scala.concurrent.Await` instead.
-   */
+    * Designates (and eventually executes) a thunk which potentially blocks the calling `java.lang.Thread`.
+    *
+    * Clients must use `scala.concurrent.blocking` or `scala.concurrent.Await` instead.
+    *
+    * In implementations of this method it is RECOMMENDED to first check if `permission` is `null` and
+    * if it is, throw an `IllegalArgumentException`.
+    *
+    * @throws IllegalArgumentException if the `permission` is `null`
+    */
   def blockOn[T](thunk: =>T)(implicit permission: CanAwait): T
 }
 
 object BlockContext {
-  private object DefaultBlockContext extends BlockContext {
-    override def blockOn[T](thunk: =>T)(implicit permission: CanAwait): T = thunk
+  private[this] final object DefaultBlockContext extends BlockContext {
+    override final def blockOn[T](thunk: =>T)(implicit permission: CanAwait): T = thunk
   }
 
   /**
-   * @return the `BlockContext` that will be used if no other is found.
-   **/
-  def defaultBlockContext: BlockContext = DefaultBlockContext
+    * The default block context will execute the supplied thunk immediately.
+    * @return the `BlockContext` that will be used if no other is found.
+    **/
+  final def defaultBlockContext: BlockContext = DefaultBlockContext
+
+  private[this] final val contextLocal = new ThreadLocal[BlockContext]()
 
-  private[this] val contextLocal = new ThreadLocal[BlockContext]()
+  private[this] final def prefer(candidate: BlockContext): BlockContext =
+    if (candidate ne null) candidate
+    else {
+      val t = Thread.currentThread
+      if (t.isInstanceOf[BlockContext]) t.asInstanceOf[BlockContext]
+      else DefaultBlockContext
+    }
 
   /**
-    @return the `BlockContext` that would be used for the current `java.lang.Thread` at this point
+   * @return the `BlockContext` that would be used for the current `java.lang.Thread` at this point
    **/
-  def current: BlockContext = contextLocal.get match {
-    case null => Thread.currentThread match {
-      case ctx: BlockContext => ctx
-      case _ => DefaultBlockContext
-    }
-    case some => some
-  }
+  final def current: BlockContext = prefer(contextLocal.get)
 
   /**
    * Installs a current `BlockContext` around executing `body`.
    **/
-  def withBlockContext[T](blockContext: BlockContext)(body: => T): T = {
+  final def withBlockContext[T](blockContext: BlockContext)(body: => T): T = {
     val old = contextLocal.get // can be null
     try {
       contextLocal.set(blockContext)
@@ -83,4 +91,14 @@ object BlockContext {
       contextLocal.set(old)
     }
   }
+
+  /**
+   * Installs the BlockContext `blockContext` around the invocation to `f` and passes in the previously installed BlockContext to `f`.
+   * @return the value produced by applying `f`
+   **/
+  final def usingBlockContext[I, T](blockContext: BlockContext)(f: BlockContext => T): T = {
+    val old = contextLocal.get // can be null
+    contextLocal.set(blockContext)
+    try f(prefer(old)) finally contextLocal.set(old)
+  }
 }