Acquire fewer locks in TaskRunner

[swankjesse]

Previously each run did this:

• acquire a lock to take a task
• acquire a lock to finish a task
• if crashed, acquire a lock to start a new thread

So to run 10 tasks without any crashes, we'd acquire the lock 20 times.

With this update, we do this:

• acquire a lock to take the first task
• acquire a lock to release task N and take task N + 1

So to run 10 tasks without any crashes, we now acquire the lock 11 times.

[swankjesse]

We also change the thread name fewer times!

[swankjesse]

I like this new order better

[yschimke]

+1 more logical

[yschimke]

Does this failure need an update?

ConnectionPoolTest > interruptStopsThread() FAILED
    org.opentest4j.AssertionFailedError: expected to be empty but was:<[Q10505]>
        at app//okhttp3.internal.connection.ConnectionPoolTest.interruptStopsThread(ConnectionPoolTest.kt:203)

[swankjesse]

I think there’s non-determinism that I need to fix in that test, as I can’t reproduce that failure locally! Lemme do that first, then rebase this on top.

[Endeavour233]

I think there’s non-determinism that I need to fix in that test, as I can’t reproduce that failure locally! Lemme do that first, then rebase this on top.

I am confused about what the interruptStopsThread is designed for. Did we expect that a TaskRunner thread(coordinator) is interrupted when it's waiting to execute a scheduled cleanupTask in cleanupQueue, and then cancel all cancellable tasks of queues in activeQueues(readyQueues + busyQueues)(in this case, cleanupTask is cancelled and removed from cleanupQueue, resulting in an empty cleanupQueue, which is then removed from readyQueues), so that we could observe that the activeQueues is empty?

If so, I think the idleAtNs of the added connection should be LONG.MAX_VALUE - 101 instead of the default value: LONG.MAX_VALUE.

 @Test fun interruptStopsThread() {
    val realTaskRunner = TaskRunner.INSTANCE
    val pool =
      factory.newConnectionPool(
        taskRunner = TaskRunner.INSTANCE,
        maxIdleConnections = 2,
      )
    // factory.newConnection(pool, routeA1, Long.MAX_VALUE - 101)?
    factory.newConnection(pool, routeA1)
    assertThat(realTaskRunner.activeQueues()).isNotEmpty()
    Thread.sleep(100)
    val threads = arrayOfNulls<Thread>(Thread.activeCount() * 2)
    Thread.enumerate(threads)
    for (t in threads) {
      if (t != null && t.name == "OkHttp TaskRunner") {
        t.interrupt()
      }
    }
    Thread.sleep(100)
    assertThat(realTaskRunner.activeQueues()).isEmpty()
  }

This default value will make the connection ignored when cleanupTask tries to figure out whether there is any connection remaining to be closed. Consequently, the cleanupTask won't be added back to cleanupQueue and the cleanupQueue won't be added back to the readyQueue after its initial execution - activeQueue is empty even if we did not interrupt threads.

// RealConnectionPool.kt      
fun closeCnnections(now: Long): Long {      
                  //...          
          if (idleAtNs < earliestEvictableIdleAtNs) { // idleAtNs = Long.MAX_VALUE earliestEvicatbleIdleAtNs = Long.MAX_VALUE
            earliestEvictableIdleAtNs = idleAtNs
            earliestEvictableConnection = connection
          }
}