IO: fix too expensive joinTo fast-path

Author: Sergey Mashkov

benchmarks/src/jmh/kotlin/benchmarks/ChannelCopyBenchmark.kt

@@ -33,6 +33,7 @@ ChannelCopyBenchmark.runBlockingAndLaunch          avgt    5   833,390 ±  14,96
 @State(Scope.Benchmark)
 @Fork(1)
 open class ChannelCopyBenchmark {
+    private val HelloWorld = "Hello, World!".toByteArray()
     private val ABC = "ABC".repeat(100).toByteArray()
     private val buffer = ByteArray(4096)
     private val ioe = IOException()
@@ -76,6 +77,26 @@ open class ChannelCopyBenchmark {
         read
     }
 
+    @Benchmark
+    fun cioChannelCopyHW() = runBlocking {
+        val pIn = ByteChannel(true)
+        val pOut = ByteChannel(true)
+
+        pOut.writeFully(HelloWorld)
+        pOut.close()
+
+        pOut.copyAndClose(pIn)
+
+        var read = 0
+        while (read < HelloWorld.size) {
+            val rc = pIn.readAvailable(buffer)
+            if (rc == -1) break
+            read += rc
+        }
+
+        read
+    }
+
     @Benchmark
     fun cioJoinToClosed() = runBlocking {
         val pIn = ByteChannel(true)
@@ -96,6 +117,53 @@ open class ChannelCopyBenchmark {
         read
     }
 
+    @Benchmark
+    fun cioJoinToClosedHW() = runBlocking {
+        val pIn = ByteChannel(true)
+        val pOut = ByteChannel(true)
+
+        pOut.writeFully(HelloWorld)
+        pOut.close()
+
+        pOut.joinTo(pIn, true)
+
+        var read = 0
+        while (read < HelloWorld.size) {
+            val rc = pIn.readAvailable(buffer)
+            if (rc == -1) break
+            read += rc
+        }
+
+        read
+    }
+
+
+    @Benchmark
+    fun cioCopyFromEmpty() = runCoroutineFast {
+        val from = ByteChannel(true)
+        val to = ByteChannel(true)
+
+        from.close()
+        from.copyAndClose(to)
+    }
+
+    @Benchmark
+    fun cioJoinFromEmpty() = runCoroutineFast {
+        val from = ByteChannel(true)
+        val to = ByteChannel(true)
+
+        from.close()
+        from.joinTo(to, true)
+    }
+
+    @Benchmark
+    fun cioJoinFromEmptyNonClosed() = runCoroutineFast(allowSuspend = true) {
+        val from = ByteChannel(true)
+        val to = ByteChannel(true)
+
+        from.joinTo(to, true) // should setup joining and suspend
+    }
+
     @Benchmark
     fun cioJoinToBeforeWrite() = runBlocking {
         val pIn = ByteChannel(true)
@@ -120,6 +188,30 @@ open class ChannelCopyBenchmark {
         read
     }
 
+    @Benchmark
+    fun cioJoinToHWBeforeWrite() = runBlocking {
+        val pIn = ByteChannel(true)
+        val pOut = ByteChannel(true)
+
+        launch(coroutineContext) {
+            pOut.joinTo(pIn, true)
+        }
+
+        yield()
+
+        pOut.writeFully(HelloWorld)
+        pOut.close()
+
+        var read = 0
+        while (read < HelloWorld.size) {
+            val rc = pIn.readAvailable(buffer)
+            if (rc == -1) break
+            read += rc
+        }
+
+        read
+    }
+
     @Benchmark
     fun cioCopyToInLaunch() = runBlocking {
         val pIn = ByteChannel(true)
@@ -145,6 +237,31 @@ open class ChannelCopyBenchmark {
         read
     }
 
+    @Benchmark
+    fun cioCopyToHWInLaunch() = runBlocking {
+        val pIn = ByteChannel(true)
+        val pOut = ByteChannel(true)
+
+        launch(coroutineContext) {
+            pOut.copyTo(pIn)
+            pIn.close()
+        }
+
+        yield()
+
+        pOut.writeFully(HelloWorld)
+        pOut.close()
+
+        var read = 0
+        while (read < HelloWorld.size) {
+            val rc = pIn.readAvailable(buffer)
+            if (rc == -1) break
+            read += rc
+        }
+
+        read
+    }
+
     @Benchmark
     fun cioJustWrite() = runBlocking {
         val c = ByteChannel()
@@ -153,7 +270,7 @@ open class ChannelCopyBenchmark {
     }
 
     @Benchmark
-    fun cioJustWriteUnintercepted() = runForSureNoSuspend {
+    fun cioJustWriteUnintercepted() = runCoroutineFast {
         val c = ByteChannel()
         c.writeFully(ABC)
         c.close(ioe)
@@ -168,7 +285,7 @@ open class ChannelCopyBenchmark {
     }
 
     @Benchmark
-    fun cioReadAndWriteUnintercepted() = runForSureNoSuspend {
+    fun cioReadAndWriteUnintercepted() = runCoroutineFast {
         val c = ByteChannel(true)
         c.writeFully(ABC)
         c.readAvailable(buffer)
@@ -188,9 +305,10 @@ open class ChannelCopyBenchmark {
         yield()
     }
 
-    private fun runForSureNoSuspend(block: suspend () -> Unit) {
+    private fun runCoroutineFast(allowSuspend: Boolean = false, block: suspend () -> Unit) {
         if (block.startCoroutineUninterceptedOrReturn(EmptyContinuation) === COROUTINE_SUSPENDED) {
-            throw IllegalStateException("Unexpected suspend")
+            if (!allowSuspend)
+                throw IllegalStateException("Unexpected suspend")
         }
     }
 

core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/ByteBufferChannel.kt

@@ -261,7 +261,7 @@ internal class ByteBufferChannel(
         if (!tryReleaseBuffer()) return false
         ensureClosedJoined(joined)
 
-        resumeReadOp(IllegalStateException("Joining is in progress"))
+        resumeReadOp { IllegalStateException("Joining is in progress") }
         resumeWriteOp() // here we don't resume it with exception because it should resume and delegate writing
 
         return true
@@ -1818,8 +1818,8 @@ internal class ByteBufferChannel(
         }
     }
 
-    private fun resumeReadOp(result: Throwable) {
-        ReadOp.getAndSet(this, null)?.resumeWithException(result)
+    private inline fun resumeReadOp(exception: () -> Throwable) {
+        ReadOp.getAndSet(this, null)?.resumeWithException(exception())
     }
 
     private fun resumeWriteOp() {