Skip to content

Commit

Permalink
work-around concurrency bug in okio.Pipe (square/okio#1412)
Browse files Browse the repository at this point in the history
  • Loading branch information
jhump committed Jan 30, 2024
1 parent 081a61e commit d0b8507
Showing 1 changed file with 19 additions and 4 deletions.
23 changes: 19 additions & 4 deletions okhttp/src/main/kotlin/com/connectrpc/okhttp/OkHttpStream.kt
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ import okio.Buffer
import okio.BufferedSink
import okio.BufferedSource
import okio.Pipe
import okio.withLock
import java.io.IOException
import java.util.concurrent.CountDownLatch

Expand Down Expand Up @@ -181,15 +182,25 @@ internal class PipeRequestBody(
) : RequestBody() {
private val pipe = Pipe(pipeMaxBufferSize)

/**
* Lock used to provide extra synchronization so that fold, write, and
* close can work correctly in the face of concurrency.
*
* See https://github.com/square/okio/issues/1412
*/
private val pipeLock = pipe.lock

/**
* Latch that signals when the pipe's sink is closed.
*/
private val closed = CountDownLatch(1)

fun write(buffer: Buffer) {
try {
pipe.sink.write(buffer, buffer.size)
pipe.sink.flush()
pipeLock.withLock {
pipe.sink.write(buffer, buffer.size)
pipe.sink.flush()
}
} catch (e: Throwable) {
close()
throw e
Expand All @@ -199,7 +210,9 @@ internal class PipeRequestBody(
override fun contentType() = contentType

override fun writeTo(sink: BufferedSink) {
pipe.fold(sink)
pipeLock.withLock {
pipe.fold(sink)
}
if (!duplex) {
// For non-duplex request bodies, okhttp3
// expects this method to return only when
Expand All @@ -214,7 +227,9 @@ internal class PipeRequestBody(

fun close() {
try {
pipe.sink.close()
pipeLock.withLock {
pipe.sink.close()
}
} catch (_: Throwable) {
// No-op
} finally {
Expand Down

0 comments on commit d0b8507

Please sign in to comment.