diff --git a/okhttp-dnsoverhttps/src/main/kotlin/okhttp3/dnsoverhttps/internal/-DnsOverHttpsCall.kt b/okhttp-dnsoverhttps/src/main/kotlin/okhttp3/dnsoverhttps/internal/-DnsOverHttpsCall.kt index 680c58e9ffd5..a2b25e505ad5 100644 --- a/okhttp-dnsoverhttps/src/main/kotlin/okhttp3/dnsoverhttps/internal/-DnsOverHttpsCall.kt +++ b/okhttp-dnsoverhttps/src/main/kotlin/okhttp3/dnsoverhttps/internal/-DnsOverHttpsCall.kt @@ -27,12 +27,34 @@ import okhttp3.Response import okhttp3.internal.OkHttpInternalApi import okhttp3.internal.testAndSet -// TODO: honor resolvePrivateAddresses, resolvePublicAddresses // TODO: in-memory caching that uses timeToLive. // TODO: honor Https.priority and Https.targetName. Create new calls! /** * Implements [Dns.Call] by making multiple HTTPS calls. + * + * Concurrency + * ----------- + * + * A few things conspire to make concurrency tricky: + * + * * Each DNS record type is queried in parallel; [onResponse] and [onFailure] may be called + * concurrently. + * * Calls to [Dns.Callback] must be serialized. + * * We don't want to use locks to guard access to [Dns.Callback] functions. + * + * Each time we receive data for the callback (in the form of records or an exception), we either + * immediately call the callback with that data (on a dispatcher thread), or queue it for the thread + * that's busy calling the callback. + * + * After calling a callback, the caller must check to see if there's more data queued to deliver, + * and deliver that also. + * + * The potentially surprising outcome of this strategy is the thread that performed the `TYPE_AAAA` + * DNS request may also deliver the `TYPE_A` records to the callback, or vice versa. + * + * If a thread is intending to call the callback, it sets [State.Running.lockHeld] to true while + * that call is executing. */ @OkHttpInternalApi internal class DnsOverHttpsCall( @@ -41,11 +63,16 @@ internal class DnsOverHttpsCall( private val canceledException: IOException?, ) : Dns.Call, Callback { - @Volatile private var canceled = false + @Volatile + private var canceled = false private val state = AtomicReference(State.Idle) override fun enqueue(callback: Dns.Callback) { - val running = State.Running(callback, calls) + val running = + State.Running( + callback = callback, + runningCalls = calls, + ) val previous = state.testAndSet(running) { it is State.Idle } check(previous is State.Idle) { @@ -65,43 +92,10 @@ internal class DnsOverHttpsCall( call: Call, e: IOException, ) { - while (true) { - val previous = - state.get() as? State.Running - ?: return // Already complete or canceled; nothing to do. - - val newRunningCalls = previous.runningCalls - call - val allFailures = - when { - canceledException != null -> listOf(canceledException) - else -> previous.delayedFailures + e - } - val next = - when { - newRunningCalls.isEmpty() -> { - State.Complete - } - - else -> { - State.Running( - callback = previous.callback, - runningCalls = newRunningCalls, - delayedFailures = allFailures, - ) - } - } - - if (!state.compareAndSet(previous, next)) continue // Lost a race; retry. - - if (next is State.Complete) { - previous.callback.onFailure( - call = this, - exceptions = allFailures, - ) - } - - return - } + updateStateAndCallCallbacks( + completedCall = call, + newException = e, + ) } override fun onResponse( @@ -112,7 +106,10 @@ internal class DnsOverHttpsCall( try { decodeResponse(response) } catch (e: IOException) { - return onFailure(call, e) + return updateStateAndCallCallbacks( + completedCall = call, + newException = e, + ) } val dnsRecords = @@ -144,16 +141,66 @@ internal class DnsOverHttpsCall( } } + updateStateAndCallCallbacks( + completedCall = call, + newRecords = dnsRecords, + ) + } + + private tailrec fun updateStateAndCallCallbacks( + completedCall: Call? = null, + newRecords: List = listOf(), + newException: IOException? = null, + lockHeldByThisThread: Boolean = false, + ) { while (true) { val previous = state.get() as? State.Running ?: return // Already complete or canceled; nothing to do. - val newRunningCalls = previous.runningCalls - call - val lastRunningCall = newRunningCalls.isEmpty() + val newRunningCalls = + when { + completedCall != null -> previous.runningCalls - completedCall + else -> previous.runningCalls + } + + val allExceptions = + when { + canceledException != null -> listOf(canceledException) + newException != null -> previous.pendingExceptions + newException + else -> previous.pendingExceptions + } + + val allRecords = + when { + newRecords.isNotEmpty() -> previous.pendingRecords + newRecords + else -> previous.pendingRecords + } + + val last = newRunningCalls.isEmpty() + val lockHeldByAnotherThread = !lockHeldByThisThread && previous.lockHeld + + // There's a few reasons why we might not call any callbacks: + // - There's no more records or failures to emit immediately + // - Another thread is already calling the callbacks. + // In such cases, hand off any new work to that other thread and be done. + if ((!last && allRecords.isEmpty()) || lockHeldByAnotherThread) { + val next = + State.Running( + callback = previous.callback, + runningCalls = newRunningCalls, + lockHeld = lockHeldByAnotherThread, + pendingRecords = allRecords, + pendingExceptions = allExceptions, + ) + if (!state.compareAndSet(previous, next)) continue // Lost a race, retry. + return + } + + // We need to call a callback. Take the lock and the records. val next = when { - lastRunningCall -> { + last -> { State.Complete } @@ -161,30 +208,35 @@ internal class DnsOverHttpsCall( State.Running( callback = previous.callback, runningCalls = newRunningCalls, - delayedFailures = previous.delayedFailures, + lockHeld = true, + pendingRecords = listOf(), + pendingExceptions = allExceptions, ) } } + if (!state.compareAndSet(previous, next)) continue // Lost a race, retry. - if (!state.compareAndSet(previous, next)) continue // Lost a race; retry. - - val emitDelayedFailures = lastRunningCall && previous.delayedFailures.isNotEmpty() - val lastEvent = lastRunningCall && !emitDelayedFailures - if (dnsRecords.isNotEmpty() || lastEvent) { + val lastAndNoExceptions = last && allExceptions.isEmpty() + if (allRecords.isNotEmpty() || lastAndNoExceptions) { previous.callback.onRecords( call = this, - last = lastEvent, - records = dnsRecords, + last = lastAndNoExceptions, + records = allRecords, ) } - if (emitDelayedFailures) { + + if (last && allExceptions.isNotEmpty()) { previous.callback.onFailure( call = this, - exceptions = previous.delayedFailures, + exceptions = allExceptions, ) } - return + // Success, attempt to release the held lock. This might also process more events if some + // were enqueued while the callback was executing. + return updateStateAndCallCallbacks( + lockHeldByThisThread = true, + ) } } @@ -204,9 +256,15 @@ internal class DnsOverHttpsCall( class Running( val callback: Dns.Callback, + val lockHeld: Boolean = false, val runningCalls: List, - val delayedFailures: List = listOf(), - ) : State + val pendingRecords: List = listOf(), + val pendingExceptions: List = listOf(), + ) : State { + init { + check(pendingRecords.isEmpty() || lockHeld) + } + } object Complete : State } diff --git a/okhttp-dnsoverhttps/src/test/java/okhttp3/dnsoverhttps/DnsOverHttpsTest.kt b/okhttp-dnsoverhttps/src/test/java/okhttp3/dnsoverhttps/DnsOverHttpsTest.kt index e9229a6396e9..c0be21e7d2b5 100644 --- a/okhttp-dnsoverhttps/src/test/java/okhttp3/dnsoverhttps/DnsOverHttpsTest.kt +++ b/okhttp-dnsoverhttps/src/test/java/okhttp3/dnsoverhttps/DnsOverHttpsTest.kt @@ -19,6 +19,7 @@ import app.cash.burst.Burst import assertk.assertThat import assertk.assertions.containsExactly import assertk.assertions.hasMessage +import assertk.assertions.hasSize import assertk.assertions.isEqualTo import assertk.assertions.isInstanceOf import assertk.assertions.isNull @@ -27,6 +28,11 @@ import java.io.EOFException import java.io.File import java.net.InetAddress import java.net.UnknownHostException +import java.util.concurrent.CompletableFuture +import java.util.concurrent.LinkedBlockingDeque +import java.util.concurrent.SynchronousQueue +import java.util.concurrent.ThreadPoolExecutor +import java.util.concurrent.TimeUnit import kotlin.reflect.KClass import kotlin.test.assertFailsWith import mockwebserver3.MockResponse @@ -93,9 +99,26 @@ class DnsOverHttpsTest( dispatcher = server.dispatcher } + private val completedRunnables = LinkedBlockingDeque() + private val executorService = + object : ThreadPoolExecutor( + 0, + Int.MAX_VALUE, + 1, + TimeUnit.SECONDS, + SynchronousQueue(), + ) { + override fun afterExecute( + r: Runnable, + t: Throwable?, + ) { + completedRunnables.add(r) + } + } + private lateinit var dns: DnsOverHttps private val dispatcher = - Dispatcher() + Dispatcher(executorService) .apply { // Force calls to run sequentially for determinism. this.maxRequestsPerHost = 1 @@ -671,6 +694,90 @@ class DnsOverHttpsTest( assertThat(dnsEvents.take()).isInstanceOf() } + @Test + fun callbackIsCalledSequentiallyWhenHttpCallsAreParallel() { + callbackIsCalledSequentially() + } + + @Test + fun callbackIsCalledSequentiallyWhenCallsFail() { + server.sequenceIndexToOverride[0] = overrideResponse("") + callbackIsCalledSequentially() + } + + /** + * We'd like to confirm that calls into [Dns.Callback] are serialized. + * + * Asserting that is awkward, so we're asserting something stronger, advised by the + * implementation: if the first call to `onRecords()` waits for the other 2 HTTP calls to + * finish, then all 3 calls will happen on the same thread. + * + * (This works because we know the other 2 HTTP calls will finish before their data is emitted.) + */ + private fun callbackIsCalledSequentially() { + assumeTrue(entryPoint == EntryPoint.NewCall) + + dns = buildLocalhost(bootstrapClient, includeIPv6 = true, includeHttps = true) + server["lysine.dev"] = + listOf( + ResourceRecord.IpAddress( + name = "lysine.dev", + timeToLive = 5, + address = InetAddress.getByName("10.20.30.40"), + ), + ResourceRecord.IpAddress( + name = "lysine.dev", + timeToLive = 5, + address = InetAddress.getByName("1:2::3:4"), + ), + ResourceRecord.Https( + name = "lysine.dev", + timeToLive = 5, + alpnIds = listOf(Protocol.HTTP_2.toString()), + ), + ) + + // Track which threads receive callbacks. + val threadsFuture = CompletableFuture>() + + dispatcher.maxRequestsPerHost = 3 + + val call = dns.newCall(Dns.Request("lysine.dev")) + call.enqueue( + object : Dns.Callback { + private val threads = mutableSetOf() + + override fun onRecords( + call: Dns.Call, + last: Boolean, + records: List, + ) { + onCallback(last) + } + + override fun onFailure( + call: Dns.Call, + e: java.io.IOException, + ) { + onCallback(true) + } + + private fun onCallback(last: Boolean) { + if (threads.isEmpty()) { + completedRunnables.take() + completedRunnables.take() + } + threads += Thread.currentThread() + if (last) { + threadsFuture.complete(threads) + } + } + }, + ) + + assertThat(threadsFuture.get()).hasSize(1) + } + private fun cacheEvents(): List> = eventRecorder .recordedEventTypes() diff --git a/okhttp-testing-support/src/main/kotlin/okhttp3/EventRecorder.kt b/okhttp-testing-support/src/main/kotlin/okhttp3/EventRecorder.kt index adaa50f994a4..85a23acf08b0 100644 --- a/okhttp-testing-support/src/main/kotlin/okhttp3/EventRecorder.kt +++ b/okhttp-testing-support/src/main/kotlin/okhttp3/EventRecorder.kt @@ -20,8 +20,8 @@ import assertk.assertions.isCloseTo import assertk.assertions.isFalse import assertk.assertions.isInstanceOf import assertk.assertions.matchesPredicate -import java.util.Deque import java.util.concurrent.ConcurrentLinkedDeque +import java.util.concurrent.LinkedBlockingQueue import java.util.concurrent.TimeUnit import okhttp3.CallEvent.CallStart import okhttp3.CallEvent.Canceled @@ -44,7 +44,7 @@ open class EventRecorder( get() = eventListenerAdapter /** Events that haven't yet been removed. */ - val eventSequence: Deque = ConcurrentLinkedDeque() + val eventSequence = LinkedBlockingQueue() /** The full set of events, used to match starts with ends. */ private val eventsForMatching = ConcurrentLinkedDeque() @@ -93,7 +93,7 @@ open class EventRecorder( eventClass: Class? = null, elapsedMs: Long = -1L, ): CallEvent { - val result = eventSequence.remove() + val result = eventSequence.take() val actualElapsedNs = result.timestampNs - (lastTimestampNs ?: result.timestampNs) lastTimestampNs = result.timestampNs diff --git a/okhttp-testing-support/src/main/kotlin/okhttp3/RecordingConnectionListener.kt b/okhttp-testing-support/src/main/kotlin/okhttp3/RecordingConnectionListener.kt index e4aa31c52997..7e897559306c 100644 --- a/okhttp-testing-support/src/main/kotlin/okhttp3/RecordingConnectionListener.kt +++ b/okhttp-testing-support/src/main/kotlin/okhttp3/RecordingConnectionListener.kt @@ -26,8 +26,7 @@ import assertk.assertions.isCloseTo import assertk.assertions.isFalse import assertk.assertions.isInstanceOf import assertk.assertions.matchesPredicate -import java.util.Deque -import java.util.concurrent.ConcurrentLinkedDeque +import java.util.concurrent.LinkedBlockingQueue import java.util.concurrent.TimeUnit import okhttp3.ConnectionEvent.NoNewExchanges import okhttp3.internal.connection.ConnectionListener @@ -42,7 +41,7 @@ internal open class RecordingConnectionListener( */ private val enforceOrder: Boolean = true, ) : ConnectionListener() { - val eventSequence: Deque = ConcurrentLinkedDeque() + val eventSequence = LinkedBlockingQueue() private val forbiddenLocks = mutableSetOf() @@ -84,7 +83,7 @@ internal open class RecordingConnectionListener( eventClass: Class? = null, elapsedMs: Long = -1L, ): ConnectionEvent { - val result = eventSequence.remove() + val result = eventSequence.take() val actualElapsedNs = result.timestampNs - (lastTimestampNs ?: result.timestampNs) lastTimestampNs = result.timestampNs diff --git a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/Dns.kt b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/Dns.kt index 51ea99b695fb..468ccb10e6e8 100644 --- a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/Dns.kt +++ b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/Dns.kt @@ -60,6 +60,12 @@ fun interface Dns { fun isCanceled(): Boolean } + /** + * Receives a stream of records from the DNS server. + * + * Calls to [onRecords] and [onFailure] may be invoked by different threads, but never in + * parallel: a call will not be made until the prior call has returned. + */ interface Callback { /** * @param last true if this is the last list of records for this address. That is a terminal