diff --git a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Connection.kt b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Connection.kt index b79401a79f64..dff38c54aa05 100644 --- a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Connection.kt +++ b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Connection.kt @@ -212,7 +212,7 @@ class Http2Connection internal constructor( out: Boolean, ): Http2Stream { check(!client) { "Client cannot push requests." } - return newStream(associatedStreamId, requestHeaders, out) + return newStream(associatedStreamId, requestHeaders, out, writeTimeoutMillis = 0L) } /** @@ -225,13 +225,28 @@ class Http2Connection internal constructor( fun newStream( requestHeaders: List
, out: Boolean, - ): Http2Stream = newStream(0, requestHeaders, out) + ): Http2Stream = newStream(0, requestHeaders, out, writeTimeoutMillis = 0L) + + /** + * Like [newStream], but bounds the blocking write of the initial HEADERS frame by + * [writeTimeoutMillis]. Without this, that write -- which happens before the returned + * [Http2Stream]'s own write timeout is configured by the caller -- can block on the underlying + * socket for as long as the OS takes to give up on an unresponsive peer (see + * square/okhttp#9237), ignoring [okhttp3.OkHttpClient.writeTimeoutMillis] entirely. + */ + @Throws(IOException::class) + internal fun newStream( + requestHeaders: List
, + out: Boolean, + writeTimeoutMillis: Long, + ): Http2Stream = newStream(0, requestHeaders, out, writeTimeoutMillis) @Throws(IOException::class) private fun newStream( associatedStreamId: Int, requestHeaders: List
, out: Boolean, + writeTimeoutMillis: Long, ): Http2Stream { val outFinished = !out val inFinished = false @@ -250,6 +265,9 @@ class Http2Connection internal constructor( streamId = nextStreamId nextStreamId += 2 stream = Http2Stream(streamId, this, outFinished, inFinished, null) + if (writeTimeoutMillis > 0L) { + stream.writeTimeout.timeout(writeTimeoutMillis, TimeUnit.MILLISECONDS) + } flushHeaders = !out || writeBytesTotal >= writeBytesMaximum || stream.writeBytesTotal >= stream.writeBytesMaximum @@ -257,19 +275,23 @@ class Http2Connection internal constructor( streams[streamId] = stream } } - if (associatedStreamId == 0) { - writer.headers(outFinished, streamId, requestHeaders) - } else { - require(!client) { "client streams shouldn't have associated stream IDs" } - // HTTP/2 has a PUSH_PROMISE frame. - writer.pushPromise(associatedStreamId, streamId, requestHeaders) + stream.guardConnectionWrite { + if (associatedStreamId == 0) { + writer.headers(outFinished, streamId, requestHeaders) + } else { + require(!client) { "client streams shouldn't have associated stream IDs" } + // HTTP/2 has a PUSH_PROMISE frame. + writer.pushPromise(associatedStreamId, streamId, requestHeaders) + } + // The write above only fills Http2Writer's internal buffer. flush() is what actually + // performs the (potentially blocking) socket write, so it must stay inside this timeout + // guard too -- see the class doc on this overload. + if (flushHeaders) { + writer.flush() + } } } - if (flushHeaders) { - writer.flush() - } - return stream } diff --git a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2ExchangeCodec.kt b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2ExchangeCodec.kt index 019c0b604331..aec3b98808c1 100644 --- a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2ExchangeCodec.kt +++ b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2ExchangeCodec.kt @@ -81,7 +81,12 @@ class Http2ExchangeCodec( val hasRequestBody = request.body != null val requestHeaders = http2HeadersList(request) - stream = http2Connection.newStream(requestHeaders, hasRequestBody) + stream = + http2Connection.newStream( + requestHeaders, + hasRequestBody, + writeTimeoutMillis = chain.writeTimeoutMillis.toLong(), + ) // We may have been asked to cancel while creating the new stream and sending the request // headers, but there was still no stream to close. if (canceled) { diff --git a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Stream.kt b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Stream.kt index 2c1b1779023a..5e715aa488df 100644 --- a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Stream.kt +++ b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Stream.kt @@ -78,6 +78,16 @@ class Http2Stream internal constructor( internal val readTimeout = StreamTimeout() internal val writeTimeout = StreamTimeout() + /** + * True while a thread is parked inside a blocking write to [connection]'s underlying socket + * (writing a HEADERS or DATA frame for this stream). Unlike a flow-control wait -- which parks + * in [Object.wait] and can always be woken by [closeLater]'s `notifyAll()` -- a thread blocked + * inside the actual socket write can only be freed by interrupting the socket itself. See + * [StreamTimeout.timedOut]. + */ + @Volatile + internal var blockedInConnectionWrite = false + /** * The reason why this stream was closed, or null if it closed normally or has not yet been * closed. @@ -615,12 +625,7 @@ class Http2Stream internal constructor( outFinished = outFinishedOnLastFrame && toWrite == sendBuffer.size } - writeTimeout.enter() - try { - connection.writeData(id, outFinished, sendBuffer, toWrite) - } finally { - writeTimeout.exitAndThrowIfTimedOut() - } + guardConnectionWrite { connection.writeData(id, outFinished, sendBuffer, toWrite) } } @Throws(IOException::class) @@ -633,7 +638,7 @@ class Http2Stream internal constructor( // TODO(jwilson): flush the connection?! while (sendBuffer.size > 0L) { emitFrame(false) - connection.flush() + guardConnectionWrite { connection.flush() } } } @@ -659,7 +664,7 @@ class Http2Stream internal constructor( while (sendBuffer.size > 0L) { emitFrame(false) } - connection.writeHeaders(id, outFinished, trailers!!.toHeaderList()) + guardConnectionWrite { connection.writeHeaders(id, outFinished, trailers!!.toHeaderList()) } } hasData -> { @@ -669,7 +674,7 @@ class Http2Stream internal constructor( } outFinished -> { - connection.writeData(id, true, null, 0L) + guardConnectionWrite { connection.writeData(id, true, null, 0L) } } } } @@ -677,7 +682,7 @@ class Http2Stream internal constructor( closed = true notifyAll() // Because doReadTimeout() may have changed. } - connection.flush() + guardConnectionWrite { connection.flush() } cancelStreamIfNecessary() } } @@ -717,6 +722,23 @@ class Http2Stream internal constructor( } } + /** + * Runs [block], which must perform a blocking write (or flush) directly against [connection]'s + * shared socket, bounded by [writeTimeout]. Unlike a flow-control wait, a thread inside [block] + * can only be freed by [StreamTimeout.timedOut] cancelling the socket -- see its doc for why. + */ + @Throws(IOException::class) + internal fun guardConnectionWrite(block: () -> T): T { + writeTimeout.enter() + try { + blockedInConnectionWrite = true + return block() + } finally { + blockedInConnectionWrite = false + writeTimeout.exitAndThrowIfTimedOut() + } + } + /** * The Okio timeout watchdog will call [timedOut] if the timeout is reached. In that case we close * the stream (asynchronously) which will notify the waiting thread. @@ -725,6 +747,17 @@ class Http2Stream internal constructor( override fun timedOut() { closeLater(ErrorCode.CANCEL) connection.sendDegradedPingLater() + + // closeLater()'s notifyAll() only wakes threads parked in Object.wait() (for example a + // stream blocked on flow control). It can't interrupt a thread that's parked inside a + // blocking write to the connection's shared socket -- such as a HEADERS or DATA frame + // write stalled by a TCP-level network blackhole (see square/okhttp#9237). Cancelling the + // socket is the only way to unblock that thread. This necessarily also terminates any + // other streams multiplexed on the connection, but a connection that can't complete a + // socket write within the caller's write timeout is unresponsive for all of them anyway. + if (this === writeTimeout && blockedInConnectionWrite) { + connection.socket.cancel() + } } override fun newTimeoutException(cause: IOException?): IOException = diff --git a/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt b/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt index a073f1d41b61..54892f73e468 100644 --- a/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt +++ b/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt @@ -43,10 +43,13 @@ import okhttp3.internal.concurrent.TaskRunner import okhttp3.internal.concurrent.notifyAll import okhttp3.internal.concurrent.wait import okhttp3.internal.concurrent.withLock +import okhttp3.internal.connection.BufferedSocket import okhttp3.internal.connection.asBufferedSocket import okio.AsyncTimeout import okio.Buffer +import okio.BufferedSink import okio.BufferedSource +import okio.Sink import okio.Source import okio.buffer import org.junit.jupiter.api.AfterEach @@ -1626,6 +1629,124 @@ class Http2ConnectionTest { assertThat(peer.takeFrame().type).isEqualTo(Http2.TYPE_RST_STREAM) } + /** + * Regression test for https://github.com/square/okhttp/issues/9237: readTimeout/writeTimeout + * did nothing for an HTTP/2 connection stuck in a real TCP-level stall (e.g. a network + * blackhole), because [Http2Stream.StreamTimeout.timedOut] only woke threads parked in a + * flow-control wait -- never a thread blocked inside an actual socket write. [BlockingSocket] + * simulates that stall: its `sink.write()` blocks until [BlockingSocket.cancel] is called, at + * which point it fails, exactly like a real socket does once closed. + * + * Before the fix, this write also had no timeout wired up at all yet: [Http2Connection.newStream] + * wrote the initial HEADERS frame before the caller had a chance to configure the returned + * stream's [Http2Stream.writeTimeout]. So on a stalled connection, even a plain GET request + * would hang forever, regardless of any configured writeTimeout. + */ + @Test fun newStreamTimesOutOnBlockedSocket() { + val socket = BlockingSocket() + val connection = + Http2Connection + .Builder(true, TaskRunner.INSTANCE) + .socket(socket, "blocked") + .build() + connection.start(sendConnectionPreface = false) + + val startNanos = System.nanoTime() + assertFailsWith { + connection.newStream(headerEntries("b", "banana"), false, writeTimeoutMillis = 500L) + } + val elapsedNanos = System.nanoTime() - startNanos + awaitWatchdogIdle() + + // Generous delta: this races the AsyncTimeout watchdog against BlockingSocket.cancel(), not + // just a notifyAll() wakeup like the flow-control timeout tests above. + assertThat(TimeUnit.NANOSECONDS.toMillis(elapsedNanos).toDouble()) + .isCloseTo(500.0, 1000.0) + } + + /** + * Regression test for https://github.com/square/okhttp/issues/9237, covering the DATA frame + * write path (an in-progress request body) rather than the initial HEADERS write covered by + * [newStreamTimesOutOnBlockedSocket]. See that test's doc for the full explanation. + */ + @Test fun dataWriteTimesOutOnBlockedSocket() { + val socket = BlockingSocket() + val connection = + Http2Connection + .Builder(true, TaskRunner.INSTANCE) + .socket(socket, "blocked") + .build() + connection.start(sendConnectionPreface = false) + + // out=true with a fresh flow-control window: newStream() returns without touching the + // blocked socket because the HEADERS frame is buffered, not flushed, in that case. + val stream = connection.newStream(headerEntries("b", "banana"), true, writeTimeoutMillis = 0L) + stream.writeTimeout().timeout(500, TimeUnit.MILLISECONDS) + + val startNanos = System.nanoTime() + assertFailsWith { + // At least EMIT_BUFFER_SIZE, so FramingSink.write() emits (and blocks on) a DATA frame + // without needing an explicit flush() call. + stream.sink.write(Buffer().write(ByteArray(16384)), 16384L) + } + val elapsedNanos = System.nanoTime() - startNanos + awaitWatchdogIdle() + + assertThat(TimeUnit.NANOSECONDS.toMillis(elapsedNanos).toDouble()) + .isCloseTo(500.0, 1000.0) + } + + /** + * A [BufferedSocket] that simulates a TCP-level network stall: the connection looks alive, but + * reads and writes block until [cancel] is called, at which point they fail -- like a real + * socket does once its underlying transport is closed out from under a blocked thread. + */ + private class BlockingSocket : BufferedSocket { + private val releaseLatch = CountDownLatch(1) + + override val source: BufferedSource = + object : Source { + override fun read( + sink: Buffer, + byteCount: Long, + ): Long = awaitRelease() + + override fun timeout() = okio.Timeout.NONE + + override fun close() {} + }.buffer() + + override val sink: BufferedSink = + object : Sink { + override fun write( + source: Buffer, + byteCount: Long, + ) { + source.skip(byteCount) + awaitRelease() + } + + override fun flush() {} + + override fun timeout() = okio.Timeout.NONE + + override fun close() {} + }.buffer() + + private fun awaitRelease(): Nothing { + try { + releaseLatch.await() + } catch (e: InterruptedException) { + throw IOException(e) + } + throw IOException("socket cancelled") + } + + override fun cancel() { + releaseLatch.countDown() + } + } + @Test fun outgoingWritesAreBatched() { // Write the mocking script. peer.sendFrame().settings(Settings())