Change TaskRunner to limit context switches.

Now we don't have to alternate between the coordinator thread and the task
thread between task runs if the task returns 0. Instead the task thread can
stay resident.

This implementation works by having task runnables that can switch from
the coordinator role (sleeping until the next task starts) and the executor
role.

square#5512

mockwebserver/src/main/java/okhttp3/mockwebserver/MockWebServer.kt

@@ -969,6 +969,9 @@ class MockWebServer : ExternalResource(), Closeable {
       val request = readRequest(stream)
       atomicRequestCount.incrementAndGet()
       requestQueue.add(request)
+      if (request.failure != null) {
+        return // Nothing to respond to.
+      }
 
       val response: MockResponse = dispatcher.dispatch(request)
 

okhttp/src/main/java/okhttp3/internal/concurrent/Task.kt

@@ -56,8 +56,6 @@ abstract class Task(
   /** Undefined unless this is in [TaskQueue.futureTasks]. */
   internal var nextExecuteNanoTime = -1L
 
-  internal var runRunnable: Runnable? = null
-
   /** Returns the delay in nanoseconds until the next execution, or -1L to not reschedule. */
   abstract fun runOnce(): Long
 
@@ -66,19 +64,5 @@ abstract class Task(
 
     check(this.queue === null) { "task is in multiple queues" }
     this.queue = queue
-
-    this.runRunnable = Runnable {
-      val currentThread = Thread.currentThread()
-      val oldName = currentThread.name
-      currentThread.name = name
-
-      var delayNanos = -1L
-      try {
-        delayNanos = runOnce()
-      } finally {
-        queue.runCompleted(this, delayNanos)
-        currentThread.name = oldName
-      }
-    }
   }
 }

okhttp/src/main/java/okhttp3/internal/concurrent/TaskQueue.kt

@@ -28,22 +28,16 @@ import java.util.concurrent.TimeUnit
 class TaskQueue internal constructor(
   private val taskRunner: TaskRunner
 ) {
-  private var shutdown = false
+  internal var shutdown = false
 
   /** This queue's currently-executing task, or null if none is currently executing. */
-  private var activeTask: Task? = null
-
-  /** True if the [activeTask] should not recur when it completes. */
-  private var cancelActiveTask = false
+  internal var activeTask: Task? = null
 
   /** Scheduled tasks ordered by [Task.nextExecuteNanoTime]. */
-  private val futureTasks = mutableListOf<Task>()
-
-  internal fun isActive(): Boolean {
-    check(Thread.holdsLock(taskRunner))
+  internal val futureTasks = mutableListOf<Task>()
 
-    return activeTask != null || futureTasks.isNotEmpty()
-  }
+  /** True if the [activeTask] should be canceled when it completes. */
+  internal var cancelActiveTask = false
 
   /**
    * Returns a snapshot of tasks currently scheduled for execution. Does not include the
@@ -87,7 +81,7 @@ class TaskQueue internal constructor(
   fun awaitIdle(delayNanos: Long): Boolean {
     val latch = CountDownLatch(1)
 
-    val task = object : Task("awaitIdle") {
+    val task = object : Task("awaitIdle", cancelable = false) {
       override fun runOnce(): Long {
         latch.countDown()
         return -1L
@@ -104,8 +98,8 @@ class TaskQueue internal constructor(
     return latch.await(delayNanos, TimeUnit.NANOSECONDS)
   }
 
-  /** Adds [task] to run in [delayNanos]. Returns true if the coordinator should run. */
-  private fun scheduleAndDecide(task: Task, delayNanos: Long): Boolean {
+  /** Adds [task] to run in [delayNanos]. Returns true if the coordinator is impacted. */
+  internal fun scheduleAndDecide(task: Task, delayNanos: Long): Boolean {
     task.initQueue(this)
 
     val now = taskRunner.backend.nanoTime()
@@ -124,7 +118,7 @@ class TaskQueue internal constructor(
     if (insertAt == -1) insertAt = futureTasks.size
     futureTasks.add(insertAt, task)
 
-    // Run the coordinator if we inserted at the front.
+    // Impact the coordinator if we inserted at the front.
     return insertAt == 0
   }
 
@@ -154,8 +148,8 @@ class TaskQueue internal constructor(
     }
   }
 
-  /** Returns true if the coordinator should run. */
-  private fun cancelAllAndDecide(): Boolean {
+  /** Returns true if the coordinator is impacted. */
+  internal fun cancelAllAndDecide(): Boolean {
     if (activeTask != null && activeTask!!.cancelable) {
       cancelActiveTask = true
     }
@@ -169,43 +163,4 @@ class TaskQueue internal constructor(
     }
     return tasksCanceled
   }
-
-  /**
-   * Posts the next available task to an executor for immediate execution.
-   *
-   * Returns the delay until the next call to this method, -1L for no further calls, or
-   * [Long.MAX_VALUE] to wait indefinitely.
-   */
-  internal fun executeReadyTask(now: Long): Long {
-    check(Thread.holdsLock(taskRunner))
-
-    if (activeTask != null) return Long.MAX_VALUE // This queue is busy.
-
-    // Check if a task is immediately ready.
-    val runTask = futureTasks.firstOrNull() ?: return -1L
-    val delayNanos = runTask.nextExecuteNanoTime - now
-    if (delayNanos <= 0) {
-      activeTask = runTask
-      futureTasks.removeAt(0)
-      taskRunner.backend.executeTask(runTask.runRunnable!!)
-      return Long.MAX_VALUE // This queue is busy until the run completes.
-    }
-
-    // Wait until the next task is ready.
-    return delayNanos
-  }
-
-  internal fun runCompleted(task: Task, delayNanos: Long) {
-    synchronized(taskRunner) {
-      check(activeTask === task)
-
-      if (delayNanos != -1L && !cancelActiveTask && !shutdown) {
-        scheduleAndDecide(task, delayNanos)
-      }
-
-      activeTask = null
-      cancelActiveTask = false
-      taskRunner.kickCoordinator(this)
-    }
-  }
 }