Further improves performance of Futures via BatchingExecutor

  * Splits the BatchingExecutor into Sync and Async Batches
  * Inlining logic in Promise.Transformation, and switching to Int
    for _xform ordinal

Author: viktorklang

src/library/scala/concurrent/BatchingExecutor.scala

@@ -50,6 +50,38 @@ private[concurrent] object BatchingExecutorStatics {
  * on the outer task completing.
  * This executor may run tasks in any order, including LIFO order.
  * There are no ordering guarantees.
+ *
+ * WARNING: Only use *EITHER* `submitAsyncBatched` OR `submitSyncBatched`!!
+ *
+ * When you implement this trait for async executors like thread pools,
+ * you're going to need to implement it something like the following:
+ *
+ * {{{
+ *  final override def submitAsync(runnable: Runnable): Unit =
+ *    super[SuperClass].execute(runnable) // To prevent reentrancy into `execute`
+ *
+ *  final override def execute(runnable: Runnable): Unit =
+ *    if (runnable.isInstanceOf[Batchable]) // Or other logic
+ *      submitAsyncBatched(runnable)
+ *    else
+ *      submitAsync(runnable)
+ *
+ *  final override def reportFailure(cause: Throwable): Unit = …
+ *  }}}
+ *
+ *  And if you want to implement if for a sync, trampolining, executor you're
+ *  going to implement it something like this:
+ *
+ * {{{
+ *  final override def submitAsync(runnable: Runnable): Unit = ()
+ *
+ *  final override def execute(runnable: Runnable): Unit =
+ *    submitSyncBatched(runnable) // You typically will want to batch everything
+ *
+ *  final override def reportFailure(cause: Throwable): Unit =
+ *    ExecutionContext.defaultReporter(cause) // Or choose something more fitting
+ * }}}
+ *
  */
 private[concurrent] trait BatchingExecutor extends Executor {
   private[this] final val _tasksLocal = new ThreadLocal[AnyRef]()
@@ -59,33 +91,7 @@ private[concurrent] trait BatchingExecutor extends Executor {
    * In order to conserve allocations, the first element in the batch is stored "unboxed" in
    * the `first` field. Subsequent Runnables are stored in the array called `other`.
   */
-  private[this] final class Batch(private[this] final val resubmitOnBlock: Boolean) extends Runnable with BlockContext with (BlockContext => Throwable) {
-    private[this] final var parentBlockContext: BlockContext = BatchingExecutorStatics.MissingParentBlockContext
-    private[this] final var first: Runnable = _
-    private[this] final var size: Int = _
-    private[this] final var other: Array[Runnable] = BatchingExecutorStatics.emptyBatchArray
-
-    def this(r: Runnable, resubmitOnBlock: Boolean) = {
-      this(resubmitOnBlock)
-      first = r
-      size = 1
-    }
-
-    private def this(first: Runnable, other: Array[Runnable], size: Int, resubmitOnBlock: Boolean) = {
-      this(resubmitOnBlock)
-      this.first = first
-      this.other = other
-      this.size = size
-    }
-
-    private[this] final def cloneAndClear(): Batch = {
-      val newBatch = new Batch(first, other, size, resubmitOnBlock)
-      this.first = null
-      this.parentBlockContext = BatchingExecutorStatics.MissingParentBlockContext
-      this.other = BatchingExecutorStatics.emptyBatchArray
-      this.size = 0
-      newBatch
-    }
+  private[this] sealed abstract class AbstractBatch protected (protected final var first: Runnable, protected final var other: Array[Runnable], protected final var size: Int) {
 
     private[this] final def ensureCapacity(curSize: Int): Array[Runnable] = {
       val curOther = this.other
@@ -111,61 +117,63 @@ private[concurrent] trait BatchingExecutor extends Executor {
       this.size = sz + 1
     }
 
-    private[this] final def runNext(): Boolean =
+    @tailrec protected final def runAll(): Unit = // TODO: Impose max limit of number of items (fairness)
       (this.size: @switch) match {
-        case 0 => false
+        case 0 =>
         case 1 =>
           val next = this.first
           this.first = null
           this.size = 0
           next.run()
-          this.size > 0// Could have changed during next.run()
+          runAll()
         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()
+          runAll()
         }
 
-    // This method runs in the delegate ExecutionContext's thread
+    protected final def runUntilFailureOrDone(): Throwable =
+      try {
+        runAll()
+        null
+      } catch {
+        case t: Throwable => t
+      }
+  }
+
+  private[this] final class AsyncBatch private(_first: Runnable, _other: Array[Runnable], _size: Int) extends AbstractBatch(_first, _other, _size) with Runnable with BlockContext with (BlockContext => Throwable) {
+    private[this] final var parentBlockContext: BlockContext = BatchingExecutorStatics.MissingParentBlockContext
+
+    final def this(runnable: Runnable) = this(runnable, BatchingExecutorStatics.emptyBatchArray, 1)
+
     override final def run(): Unit = {
-      _tasksLocal.set(this)
+      _tasksLocal.set(this) // This is later cleared in `apply` or `runWithoutResubmit`
 
-      val failure = // Only install the block context if we can resubmit on blocking
-        if (resubmitOnBlock) BlockContext.usingBlockContext(this)(this)
-        else runWithoutResubmit(runUntilFailureOrDone())
+      val f = resubmit(BlockContext.usingBlockContext(this)(this))
 
-      _tasksLocal.set(BatchingExecutorStatics.marker)
-      if (failure != null)
-        throw handleRunFailure(failure)
+      if (f != null)
+        throw f
     }
 
+    /* LOGIC FOR ASYNCHRONOUS BATCHES */
     override final def apply(prevBlockContext: BlockContext): Throwable = {
       parentBlockContext = prevBlockContext
       val failure = runUntilFailureOrDone()
       parentBlockContext = BatchingExecutorStatics.MissingParentBlockContext
+      _tasksLocal.remove()
       failure
     }
 
-    @tailrec private[this] final def runWithoutResubmit(failure: Throwable): Throwable =
-      if (failure != null && (failure.isInstanceOf[InterruptedException] || NonFatal(failure))) {
-        reportFailure(failure)
-        runWithoutResubmit(runUntilFailureOrDone())
-      } else failure
-
-    private[this] final def runUntilFailureOrDone(): Throwable =
-      try {
-        while(runNext()) {}
-
-        null
-      } catch {
-        case t: Throwable => t
-      }
-
-    private[this] final def handleRunFailure(cause: Throwable): Throwable =
-      if (resubmitOnBlock && size > 0) {
+    /* Attempts to resubmit this Batch to the underlying ExecutionContext,
+     * this only happens for Batches where `resubmitOnBlock` is `true`.
+     * Only attempt to resubmit when there are `Runnables` left to process.
+     * Note that `cause` can be `null`.
+     */
+    private[this] final def resubmit(cause: Throwable): Throwable =
+      if (this.size > 0) {
         try { submitAsync(this); cause } catch {
           case inner: Throwable =>
             if (NonFatal(inner)) {
@@ -174,50 +182,70 @@ private[concurrent] trait BatchingExecutor extends Executor {
               e
             } else inner
         }
-      } else cause
+      } else cause // TODO: consider if NonFatals should simply be `reportFailure`:ed rather than rethrown
 
-    override def blockOn[T](thunk: => T)(implicit permission: CanAwait): T = {
+    private[this] final def cloneAndClear(): AsyncBatch = {
+      val newBatch = new AsyncBatch(first, other, size)
+      this.first = null
+      this.parentBlockContext = BatchingExecutorStatics.MissingParentBlockContext
+      this.other = BatchingExecutorStatics.emptyBatchArray
+      this.size = 0
+      newBatch
+    }
+
+    override final def blockOn[T](thunk: => T)(implicit permission: CanAwait): T = {
       val pbc = parentBlockContext // Store this for later since `cloneAndClear()` will reset it
 
-      if(size > 0) // If we know there will be blocking, we don't want to keep tasks queued up because it could deadlock.
+      if(this.size > 0) // If we know there will be blocking, we don't want to keep tasks queued up because it could deadlock.
         submitAsync(cloneAndClear()) // If this throws then we have bigger problems
 
       pbc.blockOn(thunk) // Now delegate the blocking to the previous BC
     }
   }
 
-  /** Schedules the `runnable` to be executed—will only be used if `isAsync` returns `true`.
-  */
-  protected def submitAsync(runnable: Runnable): Unit
+  private[this] final class SyncBatch(runnable: Runnable) extends AbstractBatch(runnable, BatchingExecutorStatics.emptyBatchArray, 1) with Runnable {
+    @tailrec private[this] final def runWithoutResubmit(failure: Throwable): Throwable =
+      if (failure != null && (failure.isInstanceOf[InterruptedException] || NonFatal(failure))) {
+        reportFailure(failure)
+        runWithoutResubmit(runUntilFailureOrDone())
+      } else {
+        _tasksLocal.set(BatchingExecutorStatics.marker)
+        failure
+      }
 
-  /** Returns whether this `Executor` runs on the calling thread or if it `submitAsync` will execute its `Runnable`:s asynchronously.
-  */
-  protected def isAsync: Boolean = true
+    override final def run(): Unit = {
+      _tasksLocal.set(this) // This is later cleared in `runWithoutResubmit`
 
-  /** Must return `false` when `runnable` is `null`
+      val f = runWithoutResubmit(runUntilFailureOrDone())
+
+      if (f != null)
+        throw f
+    }
+  }
+
+  /** SHOULD throw a NullPointerException when `runnable` is null
   */
-  protected def batchable(runnable: Runnable): Boolean = runnable.isInstanceOf[Batchable]
+  protected def submitAsync(runnable: Runnable): Unit
 
   /** Reports that an asynchronous computation failed.
+   *  See `ExecutionContext.reportFailure(throwable: Throwable)`
   */
   protected def reportFailure(throwable: Throwable): Unit
 
-  override final def execute(runnable: Runnable): Unit = {
+  protected final def submitAsyncBatched(runnable: Runnable): Unit = {
+    val b = _tasksLocal.get
+    if (b.isInstanceOf[AsyncBatch]) b.asInstanceOf[AsyncBatch].push(runnable)
+    else submitAsync(new AsyncBatch(runnable))
+  }
+
+  protected final def submitSyncBatched(runnable: Runnable): Unit = {
     Objects.requireNonNull(runnable, "runnable is null")
-    if (isAsync) {
-      if (batchable(runnable)) {
-        val b = _tasksLocal.get
-        if (b.isInstanceOf[Batch]) b.asInstanceOf[Batch].push(runnable)
-        else submitAsync(new Batch(runnable, resubmitOnBlock = true))
-      } else submitAsync(runnable)
-    } else {
-      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
-        _tasksLocal.set(BatchingExecutorStatics.marker) // Set a marker to indicate that we are submitting synchronously
-        runnable.run()                                  // If we observe a non-null task which isn't a batch here, then allocate a batch
-        _tasksLocal.remove()                            // Since we are executing synchronously, we can clear this at the end of execution
-      } else new Batch(runnable, resubmitOnBlock = false).run()
-    }
+    val b = _tasksLocal.get
+    if (b.isInstanceOf[SyncBatch]) b.asInstanceOf[SyncBatch].push(runnable)
+    else if (b == null) {                             // If there is null in _tasksLocal, set a marker and run, inflate the Batch only if needed
+      _tasksLocal.set(BatchingExecutorStatics.marker) // Set a marker to indicate that we are submitting synchronously
+      runnable.run()                                  // If we observe a non-null task which isn't a batch here, then allocate a batch
+      _tasksLocal.remove()                            // Since we are executing synchronously, we can clear this at the end of execution
+    } else new SyncBatch(runnable).run()
   }
 }

src/library/scala/concurrent/Future.scala

@@ -868,7 +868,7 @@ object Future {
   // a side effect.
   private[concurrent] object InternalCallbackExecutor extends ExecutionContextExecutor with BatchingExecutor {
     override final def submitAsync(runnable: Runnable): Unit = reportFailure(null) // Cannot submit async
-    override final def isAsync = false
+    final override def execute(runnable: Runnable): Unit = submitSyncBatched(runnable)
     override final def reportFailure(t: Throwable): Unit =
       ExecutionContext.defaultReporter(new IllegalStateException("problem in scala.concurrent internal callback", t))
   }

src/library/scala/concurrent/impl/ExecutionContextImpl.scala

@@ -102,10 +102,13 @@ private[concurrent] object ExecutionContextImpl {
 
     new ForkJoinPool(desiredParallelism, threadFactory, threadFactory.uncaught, true) with ExecutionContextExecutorService with BatchingExecutor {
       final override def submitAsync(runnable: Runnable): Unit = super[ForkJoinPool].execute(runnable)
-      final override def isAsync = true
-      final override def batchable(runnable: Runnable): Boolean =
-          if (runnable.isInstanceOf[Promise.Transformation[_,_]]) runnable.asInstanceOf[Promise.Transformation[_,_]].benefitsFromBatching
-          else super.batchable(runnable)
+
+      final override def execute(runnable: Runnable): Unit =
+        if ((!runnable.isInstanceOf[Promise.Transformation[_,_]] || runnable.asInstanceOf[Promise.Transformation[_,_]].benefitsFromBatching) && runnable.isInstanceOf[Batchable])
+          submitAsyncBatched(runnable)
+        else
+          submitAsync(runnable)
+
       final override def reportFailure(cause: Throwable): Unit =
         getUncaughtExceptionHandler() match {
           case null =>