Addresses both performance and correctness aspects of Future

  * A regression was introduced since one of the tests in
    FutureSpec was comparing the wrong things, not noticing
    that regression led to design considerations which are
    in hindsight invalid. For that reason scala.concurrent.
    Transformation now stores its ExecutionContext for longer,
    this in order to make sure that we can use ec.reportFailure
    on NonFatal Throwables thrown when executing Future.foreach
    and Future.onComplete—this to live up to their contract, and
    to facilitate easier debugging of user code which uses Future.

  * I noticed that a typo had been introduced when changing between
    successful and failed branches when submitting Runnables to
    synchronous BatchingExecutors—this typo led to not activating
    the fast-path of that code, which yields a significant performance
    by delaying the inflation of allocating a Batch.

  * Execution of batched Runnables for synchronous BatchingExecutors
    has now been slightly improved due to reducing reads.

  * Submitting Runnables to BatchingExectuor now null-checks all
    Runnables, before would have cases where checks were not performed.

  * Improves performance of completeWith by avoiding to allocate the
    function which would attempt to do the actual completion.

Author: viktorklang

.gitignore

@@ -50,5 +50,7 @@
 /project/project/target/
 /project/project/project/target/
 /test/macro-annot/target/
+/test/files/target/
+/test/target/
 /build-sbt/
 local.sbt

src/library/scala/concurrent/BatchingExecutor.scala

@@ -15,7 +15,7 @@ package scala.concurrent
 import java.util.concurrent.Executor
 import java.util.Objects
 import scala.util.control.NonFatal
-import scala.annotation.tailrec
+import scala.annotation.{switch, tailrec}
 
 /**
  * Marker trait to indicate that a Runnable is Batchable by BatchingExecutors
@@ -24,7 +24,7 @@ trait Batchable {
   self: Runnable =>
 }
 
-private[concurrent] final object BatchingExecutorStatics {
+private[concurrent] object BatchingExecutorStatics {
   final val emptyBatchArray: Array[Runnable] = new Array[Runnable](0)
   final val marker = ""
   final object MissingParentBlockContext extends BlockContext {
@@ -111,21 +111,23 @@ private[concurrent] trait BatchingExecutor extends Executor {
       this.size = sz + 1
     }
 
-    final def pop(): Runnable = {
-      val sz = this.size
-      if (sz < 2) {
-        val ret = this.first
-        this.first = null
-        this.size = 0
-        ret
-      } else {
-        val o = this.other
-        val ret = o(sz - 2)
-        o(sz - 2) = null
-        this.size = sz - 1
-        ret
-      }
-    }
+    private[this] final def runNext(): Boolean =
+      (this.size: @switch) match {
+        case 0 => false
+        case 1 =>
+          val next = this.first
+          this.first = null
+          this.size = 0
+          next.run()
+          this.size > 0// Could have changed during next.run()
+        case sz =>
+          val o = this.other
+          val next = o(sz - 2)
+          o(sz - 2) = null
+          this.size = sz - 1// Important to update prior to `r.run()`
+          next.run()
+          this.size > 0// Could have changed during next.run()
+        }
 
     // This method runs in the delegate ExecutionContext's thread
     override final def run(): Unit = {
@@ -135,7 +137,7 @@ private[concurrent] trait BatchingExecutor extends Executor {
         if (resubmitOnBlock) BlockContext.usingBlockContext(this)(this)
         else runWithoutResubmit(runUntilFailureOrDone())
 
-      _tasksLocal.remove()
+      _tasksLocal.set(BatchingExecutorStatics.marker)
       if (failure != null)
         throw handleRunFailure(failure)
     }
@@ -155,8 +157,7 @@ private[concurrent] trait BatchingExecutor extends Executor {
 
     private[this] final def runUntilFailureOrDone(): Throwable =
       try {
-        while(size > 0)
-          pop().run()
+        while(runNext()) {}
 
         null
       } catch {
@@ -201,16 +202,15 @@ private[concurrent] trait BatchingExecutor extends Executor {
   */
   protected def reportFailure(throwable: Throwable): Unit
 
-  override final def execute(runnable: Runnable): Unit =
+  override final def execute(runnable: Runnable): Unit = {
+    Objects.requireNonNull(runnable, "runnable is null")
     if (isAsync) {
       if (batchable(runnable)) {
-        // We don't check if `runnable` is null here because if it is, it will be sent to `submitAsync` which will check that in the implementation(s)
         val b = _tasksLocal.get
         if (b.isInstanceOf[Batch]) b.asInstanceOf[Batch].push(runnable)
         else submitAsync(new Batch(runnable, resubmitOnBlock = true))
       } else submitAsync(runnable)
     } else {
-      Objects.requireNonNull(runnable, "runnable is null")
       val b = _tasksLocal.get
       if (b.isInstanceOf[Batch]) b.asInstanceOf[Batch].push(runnable)
       else if (b == null) {                             // If there is null in _tasksLocal, set a marker and run, inflate the Batch only if needed
@@ -219,4 +219,5 @@ private[concurrent] trait BatchingExecutor extends Executor {
         _tasksLocal.remove()                            // Since we are executing synchronously, we can clear this at the end of execution
       } else new Batch(runnable, resubmitOnBlock = false).run()
     }
+  }
 }

src/library/scala/concurrent/impl/Promise.scala

@@ -101,7 +101,7 @@ private[concurrent] object Promise {
       }
 
   // Left non-final to enable addition of extra fields by Java/Scala converters in scala-java8-compat.
-  class DefaultPromise[T] private[this] (initial: AnyRef) extends AtomicReference[AnyRef](initial) with scala.concurrent.Promise[T] with scala.concurrent.Future[T] {
+  class DefaultPromise[T] private[this] (initial: AnyRef) extends AtomicReference[AnyRef](initial) with scala.concurrent.Promise[T] with scala.concurrent.Future[T] with (Try[T] => Unit) {
     /**
      * Constructs a new, completed, Promise.
      */
@@ -112,6 +112,13 @@ private[concurrent] object Promise {
      */
     final def this() = this(Noop: AnyRef)
 
+    /**
+     * WARNING: the `resolved` value needs to have been pre-resolved using `resolve()`
+     * INTERNAL API
+     */
+    override final def apply(resolved: Try[T]): Unit =
+      tryComplete0(get(), resolved)
+
     /**
      * Returns the associated `Future` with this `Promise`
      */
@@ -248,11 +255,13 @@ private[concurrent] object Promise {
       } else /* if(state.isInstanceOf[Try[T]]) */ false
 
     override final def completeWith(other: Future[T]): this.type = {
-      val state = get()
-      if ((other ne this) && !state.isInstanceOf[Try[T]]) {
-        val resolved = if (other.isInstanceOf[DefaultPromise[T]]) other.asInstanceOf[DefaultPromise[T]].value0 else null
-        if (resolved ne null) tryComplete0(state, resolved)
-        else super.completeWith(other)
+      if (other ne this) {
+        val state = get()
+        if (!state.isInstanceOf[Try[T]]) {
+          val resolved = if (other.isInstanceOf[DefaultPromise[T]]) other.asInstanceOf[DefaultPromise[T]].value0 else other.value.orNull
+          if (resolved ne null) tryComplete0(state, resolved)
+          else other.onComplete(this)(InternalCallbackExecutor)
+        }
       }
 
       this
@@ -362,10 +371,11 @@ private[concurrent] object Promise {
    **/
   final class Transformation[-F, T] private[this] (
     private[this] final var _fun: Any => Any,
-    private[this] final var _arg: AnyRef,
+    private[this] final var _ec: ExecutionContext,
+    private[this] final var _arg: Try[F],
     private[this] final val _xform: Byte
   ) extends DefaultPromise[T]() with Callbacks[F] with Runnable with Batchable with OnCompleteRunnable {
-    final def this(xform: Int, f: _ => _, ec: ExecutionContext) = this(f.asInstanceOf[Any => Any], ec.prepare(): AnyRef, xform.toByte)
+    final def this(xform: Int, f: _ => _, ec: ExecutionContext) = this(f.asInstanceOf[Any => Any], ec.prepare(), null, xform.toByte)
 
     final def benefitsFromBatching: Boolean = _xform != Xform_onComplete && _xform != Xform_foreach
 
@@ -374,9 +384,9 @@ private[concurrent] object Promise {
     // Invariant: _arg is `ExecutionContext`, and non-null. `this` ne Noop.
     // requireNonNull(resolved) will hold as guarded by `resolve`
     final def submitWithValue(resolved: Try[F]): this.type = {
-      val e = _arg.asInstanceOf[ExecutionContext]
       _arg = resolved
-      try e.execute(this) /* Safe publication of _arg and _fun */
+      val e = _ec
+      try e.execute(this) /* Safe publication of _arg, _fun, _ec */
       catch {
         case t: Throwable => handleFailure(t, e)
       }
@@ -387,81 +397,77 @@ private[concurrent] object Promise {
     private[this] final def handleFailure(t: Throwable, e: ExecutionContext): Unit = {
       _fun = null // allow to GC
       _arg = null // see above
+      _ec  = null // see above again
       val wasInterrupted = t.isInstanceOf[InterruptedException]
       if (wasInterrupted || NonFatal(t)) {
         val completed = tryComplete0(get(), resolve(Failure(t)))
         if (completed && wasInterrupted) Thread.currentThread.interrupt()
-        if (!completed && (e ne null)) e.reportFailure(t)
+
+        // Report or rethrow failures which are unlikely to otherwise be noticed
+        if (_xform == Xform_foreach || _xform == Xform_onComplete || !completed)
+          e.reportFailure(t)
       } else throw t
     }
 
+    @inline private[this] final def completeWithLink(f: Future[T]): Try[T] = {
+      if (f.isInstanceOf[DefaultPromise[T]])
+        f.asInstanceOf[DefaultPromise[T]].linkRootOf(this, null)
+      else
+        completeWith(f)
+
+      null
+    }
+
     // Gets invoked by the ExecutionContext, when we have a value to transform.
-    // Invariant: if (_arg.isInstanceOf[Try[F]] && (_fun ne null))
-    override final def run(): Unit =
+    override final def run(): Unit = {
+      val v   = _arg
+      val fun = _fun
+      val ec  = _ec
       try {
-        val v = _arg.asInstanceOf[Try[F]]
-        (_xform.toInt: @switch) match {
-          case Xform_noop          => doAbort(v)
-          case Xform_map           => doMap(v)
-          case Xform_flatMap       => doFlatMap(v)
-          case Xform_transform     => doTransform(v)
-          case Xform_transformWith => doTransformWith(v)
-          case Xform_foreach       => v.foreach(_fun)
-          case Xform_onComplete    => _fun(v)
-          case Xform_recover       => doRecover(v)
-          case Xform_recoverWith   => doRecoverWith(v)
-          case Xform_filter        => doFilter(v)
-          case Xform_collect       => doCollect(v)
-          case _                   => doAbort(v)
-        }
+        val resolvedResult: Try[T] =
+          (_xform.toInt: @switch) match {
+            case Xform_noop          =>
+              null
+            case Xform_map           =>
+              if (v.isInstanceOf[Success[F]]) Success(fun(v.asInstanceOf[Success[F]].value).asInstanceOf[T]) else v.asInstanceOf[Failure[T]]
+            case Xform_flatMap       =>
+              if (v.isInstanceOf[Success[F]]) completeWithLink(fun(v.asInstanceOf[Success[F]].value).asInstanceOf[Future[T]])
+              else v.asInstanceOf[Failure[T]] // Already resolved
+            case Xform_transform     =>
+              resolve(fun(v).asInstanceOf[Try[T]])
+            case Xform_transformWith =>
+              completeWithLink(fun(v).asInstanceOf[Future[T]])
+            case Xform_foreach       =>
+              if (v.isInstanceOf[Success[F]]) fun(v.asInstanceOf[Success[F]].value)
+              null
+            case Xform_onComplete    =>
+              fun(v)
+              null
+            case Xform_recover       =>
+              resolve(v.recover(fun.asInstanceOf[PartialFunction[Throwable, F]]).asInstanceOf[Try[T]]) //recover F=:=T
+            case Xform_recoverWith   =>
+              if (v.isInstanceOf[Failure[F]]) {
+                val f = fun.asInstanceOf[PartialFunction[Throwable, Future[T]]].applyOrElse(v.asInstanceOf[Failure[F]].exception, Future.recoverWithFailed)
+                if (f ne Future.recoverWithFailedMarker) completeWithLink(f)
+                else v.asInstanceOf[Failure[T]]
+              } else v.asInstanceOf[Success[T]]
+            case Xform_filter        =>
+              if (v.isInstanceOf[Failure[F]] || fun.asInstanceOf[F => Boolean](v.asInstanceOf[Success[F]].value)) v.asInstanceOf[Try[T]]
+              else Future.filterFailure // Safe for unresolved completes
+            case Xform_collect       =>
+              if (v.isInstanceOf[Success[F]]) Success(fun.asInstanceOf[PartialFunction[F, T]].applyOrElse(v.asInstanceOf[Success[F]].value, Future.collectFailed))
+              else v.asInstanceOf[Failure[T]] // Already resolved
+            case _                   =>
+              Failure(new IllegalStateException("BUG: encountered transformation promise with illegal type: " + _xform))
+          }
+        if (resolvedResult ne null)
+          tryComplete0(get(), resolvedResult)
         _fun = null // allow to GC
         _arg = null // see above
+        _ec  = null // see above
       } catch {
-        case t: Throwable => handleFailure(t, null)
+        case t: Throwable => handleFailure(t, ec)
       }
-
-    private[this] final def doMap(v: Try[F]): Unit = { tryComplete0(get(), resolve(v.map(_fun.asInstanceOf[F => T]))) }
-
-    private[this] final def doFlatMap(v: Try[F]): Unit =
-      if (v.isInstanceOf[Success[F]]) {
-        val f = _fun(v.asInstanceOf[Success[F]].value)
-        if(f.isInstanceOf[DefaultPromise[T]]) f.asInstanceOf[DefaultPromise[T]].linkRootOf(this, null)
-        else completeWith(f.asInstanceOf[Future[T]])
-      } else tryComplete0(get(), v.asInstanceOf[Try[T]]) // Already resolved
-
-    private[this] final def doTransform(v: Try[F]): Unit = tryComplete0(get(), resolve(_fun(v).asInstanceOf[Try[T]]))
-
-    private[this] final def doTransformWith(v: Try[F]): Unit = {
-      val f = _fun(v)
-      if(f.isInstanceOf[DefaultPromise[T]]) f.asInstanceOf[DefaultPromise[T]].linkRootOf(this, null)
-      else completeWith(f.asInstanceOf[Future[T]])
     }
-
-    private[this] final def doRecover(v: Try[F]): Unit =
-      tryComplete0(get(), resolve(v.recover(_fun.asInstanceOf[PartialFunction[Throwable, F]]).asInstanceOf[Try[T]])) //recover F=:=T
-
-    private[this] final def doRecoverWith(v: Try[F]): Unit = //recoverWith F=:=T
-      if (v.isInstanceOf[Failure[F]]) {
-        val f = _fun.asInstanceOf[PartialFunction[Throwable, Future[T]]].applyOrElse(v.asInstanceOf[Failure[F]].exception, Future.recoverWithFailed)
-        if (f ne Future.recoverWithFailedMarker) {
-           if(f.isInstanceOf[DefaultPromise[T]]) f.asInstanceOf[DefaultPromise[T]].linkRootOf(this, null)
-           else completeWith(f)
-        } else tryComplete0(get(), v.asInstanceOf[Failure[T]])
-      } else tryComplete0(get(), v.asInstanceOf[Try[T]])
-
-    private[this] final def doFilter(v: Try[F]): Unit =
-      tryComplete0(get(),
-        if (v.isInstanceOf[Failure[F]] || _fun.asInstanceOf[F => Boolean](v.asInstanceOf[Success[F]].value)) v.asInstanceOf[Try[T]]
-        else Future.filterFailure // Safe for unresolved completes
-      )
-
-    private[this] final def doCollect(v: Try[F]): Unit =
-      tryComplete0(get(),
-        if (v.isInstanceOf[Success[F]]) Success(_fun.asInstanceOf[PartialFunction[F, T]].applyOrElse(v.asInstanceOf[Success[F]].value, Future.collectFailed))
-        else v.asInstanceOf[Try[T]] // Already resolved
-      )
-
-    private[this] final def doAbort(v: Try[F]): Unit =
-      tryComplete0(get(), Failure(new IllegalStateException("BUG: encountered transformation promise with illegal type: " + _xform)))
   }
 }

test/files/jvm/future-spec/FutureTests.scala

@@ -37,7 +37,7 @@ class FutureTests extends MinimalScalaTest {
   "A future with custom ExecutionContext" should {
     "shouldHandleThrowables" in {
       val ms = new concurrent.TrieMap[Throwable, Unit]
-      implicit val ec = scala.concurrent.ExecutionContext.fromExecutor(new java.util.concurrent.ForkJoinPool(), {
+      implicit val ec = scala.concurrent.ExecutionContext.fromExecutor(new java.util.concurrent.ForkJoinPool(1), {
         t =>
         ms.addOne((t, ()))
       })
@@ -57,28 +57,23 @@ class FutureTests extends MinimalScalaTest {
         Await.ready(latch, 5 seconds)
         "success"
       }
-      val f3 = f2 map { s => s.toUpperCase }
 
       f2 foreach { _ => throw new ThrowableTest("dispatcher foreach") }
-      f2 onComplete { case Success(_) => throw new ThrowableTest("dispatcher receive"); case _ => }
+      f2 onComplete { case Success(_) => throw new ThrowableTest("dispatcher onComplete"); case _ => }
 
       latch.open()
 
       Await.result(f2, defaultTimeout) mustBe ("success")
 
       f2 foreach { _ => throw new ThrowableTest("current thread foreach") }
-      f2 onComplete { case Success(_) => throw new ThrowableTest("current thread receive"); case _ => }
+      f2 onComplete { case Success(_) => throw new ThrowableTest("current thread onComplete"); case _ => }
 
-      Await.result(f3, defaultTimeout) mustBe ("SUCCESS")
+      Await.result(f2 map { s => s.toUpperCase }, defaultTimeout) mustBe ("SUCCESS")
 
-      val waiting = Future {
-        Thread.sleep(1000)
-      }
-      Await.ready(waiting, 4000 millis)
-
-      if (ms.size != 4)
-        assert(ms.size != 4, "Expected 4 throwables, found: " + ms)
-      //FIXME should check
+      ms.size mustBe 4
+      val msgs = ms.keysIterator.map(_.getMessage).toSet
+      val expectedMsgs = Set("dispatcher foreach", "dispatcher onComplete", "current thread foreach", "current thread onComplete")
+      msgs mustBe expectedMsgs
     }
   }