Start publishing dispatcher queue events (#9111)

* Start publishing dispatcher queue events

* Update okhttp/src/commonJvmAndroid/kotlin/okhttp3/Dispatcher.kt

---------

Co-authored-by: Jesse Wilson <jwilson@squareup.com>

Author: swankjesse

okhttp-testing-support/src/main/kotlin/okhttp3/RecordingEventListener.kt

@@ -69,8 +69,12 @@ open class RecordingEventListener(
    */
   private val enforceOrder: Boolean = true,
 ) : EventListener() {
+  /** Events that haven't yet been removed. */
   val eventSequence: Deque<CallEvent> = ConcurrentLinkedDeque()
 
+  /** The full set of events, used to match starts with ends. */
+  private val eventsForMatching = ConcurrentLinkedDeque<CallEvent>()
+
   private val forbiddenLocks = mutableListOf<Any>()
 
   /** The timestamp of the last taken event, used to measure elapsed time between events. */
@@ -151,14 +155,15 @@ open class RecordingEventListener(
       checkForStartEvent(e)
     }
 
+    eventsForMatching.offer(e)
     eventSequence.offer(e)
   }
 
   private fun checkForStartEvent(e: CallEvent) {
-    if (eventSequence.isEmpty()) {
+    if (eventsForMatching.isEmpty()) {
       assertThat(e).matchesPredicate { it is CallStart || it is Canceled }
     } else {
-      eventSequence.forEach loop@{
+      eventsForMatching.forEach loop@{
         when (e.closes(it)) {
           null -> return // no open event
           true -> return // found open event

okhttp/src/commonJvmAndroid/kotlin/okhttp3/Dispatcher.kt

@@ -170,27 +170,21 @@ class Dispatcher() {
 
     // Actions to take outside the synchronized block.
     class Effects(
-      val callsToExecute: List<AsyncCall> = listOf(),
-      val callsToReject: List<AsyncCall> = listOf(),
+      val callsToExecute: List<AsyncCall>,
       val idleCallbackToRun: Runnable?,
     )
 
     val effects =
       synchronized(this) {
-        var becameIdle = false
         if (finishedCall != null) {
           check(runningSyncCalls.remove(finishedCall)) { "Call wasn't in-flight!" }
-          becameIdle = runningSyncCalls.isEmpty()
         }
 
         if (finishedAsyncCall != null) {
           finishedAsyncCall.callsPerHost.decrementAndGet()
           check(runningAsyncCalls.remove(finishedAsyncCall)) { "Call wasn't in-flight!" }
-          becameIdle = runningAsyncCalls.isEmpty()
         }
 
-        val idleCallbackToRun = if (becameIdle) idleCallback else null
-
         if (enqueuedCall != null) {
           readyAsyncCalls.add(enqueuedCall)
 
@@ -202,9 +196,15 @@ class Dispatcher() {
           }
         }
 
+        val becameIdle =
+          (finishedCall != null || finishedAsyncCall != null) &&
+            (executorIsShutdown || runningAsyncCalls.isEmpty()) &&
+            runningSyncCalls.isEmpty()
+        val idleCallbackToRun = if (becameIdle) idleCallback else null
+
         if (executorIsShutdown) {
           return@synchronized Effects(
-            callsToReject =
+            callsToExecute =
               readyAsyncCalls
                 .toList()
                 .also { readyAsyncCalls.clear() },
@@ -233,12 +233,28 @@ class Dispatcher() {
         )
       }
 
-    for (i in 0 until effects.callsToReject.size) {
-      effects.callsToReject[i].failRejected()
-    }
+    var callDispatcherQueueStart = true
 
     for (i in 0 until effects.callsToExecute.size) {
-      effects.callsToExecute[i].executeOn(executorService)
+      val call = effects.callsToExecute[i]
+
+      // If the newly-enqueued call is already out, skip its dispatcher queue events. We only
+      // publish those events for calls that have to wait.
+      if (call === enqueuedCall) {
+        callDispatcherQueueStart = false
+      } else {
+        call.call.eventListener.dispatcherQueueEnd(call.call, this)
+      }
+
+      if (executorIsShutdown) {
+        call.failRejected()
+      } else {
+        call.executeOn(executorService)
+      }
+    }
+
+    if (callDispatcherQueueStart && enqueuedCall != null) {
+      enqueuedCall.call.eventListener.dispatcherQueueStart(enqueuedCall.call, this)
     }
 
     effects.idleCallbackToRun?.run()

okhttp/src/jvmTest/kotlin/okhttp3/DispatcherTest.kt

@@ -21,14 +21,19 @@ import assertk.assertions.containsExactlyInAnyOrder
 import assertk.assertions.isEmpty
 import assertk.assertions.isEqualTo
 import assertk.assertions.isFalse
+import assertk.assertions.isInstanceOf
 import assertk.assertions.isTrue
+import assertk.assertions.none
 import java.io.IOException
 import java.io.InterruptedIOException
 import java.net.UnknownHostException
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.test.assertFailsWith
+import okhttp3.CallEvent.DispatcherQueueEnd
+import okhttp3.CallEvent.DispatcherQueueStart
+import okhttp3.HttpUrl.Companion.toHttpUrl
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Tag
 import org.junit.jupiter.api.Test
@@ -78,6 +83,9 @@ class DispatcherTest {
   fun enqueuedJobsRunImmediately() {
     client.newCall(newRequest("http://a/1")).enqueue(callback)
     executor.assertJobs("http://a/1")
+
+    assertThat(listener.eventSequence).none { it.isInstanceOf<DispatcherQueueStart>() }
+    assertThat(listener.eventSequence).none { it.isInstanceOf<DispatcherQueueEnd>() }
   }
 
   @Test
@@ -88,6 +96,10 @@ class DispatcherTest {
     client.newCall(newRequest("http://b/1")).enqueue(callback)
     client.newCall(newRequest("http://b/2")).enqueue(callback)
     executor.assertJobs("http://a/1", "http://a/2", "http://b/1")
+
+    val dispatcherQueueStart = listener.removeUpToEvent<DispatcherQueueStart>()
+    assertThat(dispatcherQueueStart.call.request().url).isEqualTo("http://b/2".toHttpUrl())
+    assertThat(listener.eventSequence).none { it.isInstanceOf<DispatcherQueueEnd>() }
   }
 
   @Test
@@ -97,6 +109,10 @@ class DispatcherTest {
     client.newCall(newRequest("http://a/2")).enqueue(callback)
     client.newCall(newRequest("http://a/3")).enqueue(callback)
     executor.assertJobs("http://a/1", "http://a/2")
+
+    val dispatcherQueueStart = listener.removeUpToEvent<DispatcherQueueStart>()
+    assertThat(dispatcherQueueStart.call.request().url).isEqualTo("http://a/3".toHttpUrl())
+    assertThat(listener.eventSequence).none { it.isInstanceOf<DispatcherQueueEnd>() }
   }
 
   @Test
@@ -116,8 +132,23 @@ class DispatcherTest {
     client.newCall(newRequest("http://c/1")).enqueue(callback)
     client.newCall(newRequest("http://a/2")).enqueue(callback)
     client.newCall(newRequest("http://b/2")).enqueue(callback)
+
+    val dispatcherQueueStartC1 = listener.removeUpToEvent<DispatcherQueueStart>()
+    assertThat(dispatcherQueueStartC1.call.request().url).isEqualTo("http://c/1".toHttpUrl())
+    val dispatcherQueueStartA2 = listener.removeUpToEvent<DispatcherQueueStart>()
+    assertThat(dispatcherQueueStartA2.call.request().url).isEqualTo("http://a/2".toHttpUrl())
+    val dispatcherQueueStartB2 = listener.removeUpToEvent<DispatcherQueueStart>()
+    assertThat(dispatcherQueueStartB2.call.request().url).isEqualTo("http://b/2".toHttpUrl())
+    assertThat(listener.eventSequence).none { it.isInstanceOf<DispatcherQueueEnd>() }
+
     dispatcher.maxRequests = 4
     executor.assertJobs("http://a/1", "http://b/1", "http://c/1", "http://a/2")
+
+    val dispatcherQueueEndC1 = listener.removeUpToEvent<DispatcherQueueEnd>()
+    assertThat(dispatcherQueueEndC1.call.request().url).isEqualTo("http://c/1".toHttpUrl())
+    val dispatcherQueueEndA2 = listener.removeUpToEvent<DispatcherQueueEnd>()
+    assertThat(dispatcherQueueEndA2.call.request().url).isEqualTo("http://a/2".toHttpUrl())
+    assertThat(listener.eventSequence).none { it.isInstanceOf<DispatcherQueueEnd>() }
   }
 
   @Test
@@ -128,8 +159,23 @@ class DispatcherTest {
     client.newCall(newRequest("http://a/3")).enqueue(callback)
     client.newCall(newRequest("http://a/4")).enqueue(callback)
     client.newCall(newRequest("http://a/5")).enqueue(callback)
+
+    val dispatcherQueueStartA3 = listener.removeUpToEvent<DispatcherQueueStart>()
+    assertThat(dispatcherQueueStartA3.call.request().url).isEqualTo("http://a/3".toHttpUrl())
+    val dispatcherQueueStartA4 = listener.removeUpToEvent<DispatcherQueueStart>()
+    assertThat(dispatcherQueueStartA4.call.request().url).isEqualTo("http://a/4".toHttpUrl())
+    val dispatcherQueueStartA5 = listener.removeUpToEvent<DispatcherQueueStart>()
+    assertThat(dispatcherQueueStartA5.call.request().url).isEqualTo("http://a/5".toHttpUrl())
+    assertThat(listener.eventSequence).none { it.isInstanceOf<DispatcherQueueEnd>() }
+
     dispatcher.maxRequestsPerHost = 4
     executor.assertJobs("http://a/1", "http://a/2", "http://a/3", "http://a/4")
+
+    val dispatcherQueueEndA3 = listener.removeUpToEvent<DispatcherQueueEnd>()
+    assertThat(dispatcherQueueEndA3.call.request().url).isEqualTo("http://a/3".toHttpUrl())
+    val dispatcherQueueEndA4 = listener.removeUpToEvent<DispatcherQueueEnd>()
+    assertThat(dispatcherQueueEndA4.call.request().url).isEqualTo("http://a/4".toHttpUrl())
+    assertThat(listener.eventSequence).none { it.isInstanceOf<DispatcherQueueEnd>() }
   }
 
   @Test