IO: fix error propagation in blocking adapter

Author: Sergey Mashkov

core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/jvm/javaio/Blocking.kt

@@ -67,16 +67,24 @@ private class OutputAdapter(parent: Job?, private val channel: ByteWriteChannel)
             try {
                 while (true) {
                     val task = rendezvous(0)
-                    if (task === CloseToken) break
-                    if (task === FlushToken) channel.flush()
+                    if (task === CloseToken) {
+                        break
+                    }
+                    else if (task === FlushToken) {
+                        channel.flush()
+                        channel.closedCause?.let { throw it }
+                    }
                     else if (task is ByteArray) channel.writeFully(task, offset, length)
                 }
             } catch (t: Throwable) {
                 if (t !is CancellationException) {
                     channel.close(t)
                 }
+                throw t
             } finally {
-                channel.close()
+                if (!channel.close()) {
+                    channel.closedCause?.let { throw it }
+                }
             }
         }
     }
@@ -102,8 +110,12 @@ private class OutputAdapter(parent: Job?, private val channel: ByteWriteChannel)
 
     @Synchronized
     override fun close() {
-        loop.submitAndAwait(CloseToken)
-        loop.shutdown()
+        try {
+            loop.submitAndAwait(CloseToken)
+            loop.shutdown()
+        } catch (t: Throwable) {
+            throw IOException(t)
+        }
     }
 }
 
@@ -224,6 +236,12 @@ private abstract class BlockingAdapter(val parent: Job? = null) {
             LockSupport.park()
         }
 
+        state.value.let { state ->
+            if (state is Throwable) {
+                throw state
+            }
+        }
+
         return result.value
     }
 

core/kotlinx-coroutines-io/src/test/kotlin/kotlinx/coroutines/experimental/io/JavaIOTest.kt

@@ -149,7 +149,7 @@ class JavaIOTest : TestBase() {
     @Test
     fun testPipedALot() = runBlocking {
         val exec = newFixedThreadPoolContext(2, "blocking-io")
-        val numberOfLines = 10000
+        val numberOfLines = 1000 * stressTestMultiplier
         val pipe = Pipe.open()
 
         val channel1 = ByteChannel()
@@ -235,7 +235,7 @@ class JavaIOTest : TestBase() {
             val data = ByteArray(4096)
             Random().nextBytes(data)
 
-            repeat(10000) {
+            repeat(100 * stressTestMultiplier * stressTestMultiplierSqrt) {
                 val channel = ByteChannel(false)
                 launch(exec) {
                     for (i in 1..count) {
@@ -276,4 +276,82 @@ class JavaIOTest : TestBase() {
             }
         }
     }
+
+    @Test
+    fun testOutputAdapterExceptionFromWrite() = runTest {
+        val channel = ByteChannel(true)
+
+        val output = channel.toOutputStream()
+
+        launch(coroutineContext) {
+            channel.cancel()
+        }
+
+        yield()
+
+        try {
+            output.write(1)
+            fail("write() should fail")
+        } catch (expected: CancellationException) {
+        }
+    }
+
+    @Test
+    fun testOutputAdapterExceptionFromClose() = runTest {
+        val channel = ByteChannel(true)
+
+        val output = channel.toOutputStream()
+
+        launch(coroutineContext) {
+            channel.cancel()
+        }
+
+        yield()
+
+        try {
+            output.close()
+            fail("close() should fail")
+        } catch (expected: IOException) {
+        }
+    }
+
+    @Test
+    fun testOutputAdapterExceptionFromFlush() = runTest {
+        val channel = ByteChannel(true)
+
+        val output = channel.toOutputStream()
+
+        launch(coroutineContext) {
+            channel.cancel()
+        }
+
+        yield()
+
+        try {
+            output.flush()
+            fail("flush() should fail")
+        } catch (expected: CancellationException) {
+        }
+    }
+
+    @Test
+    fun testOutputAdapterExceptionFromUse() = runTest {
+        val channel = ByteChannel(true)
+
+        val output = channel.toOutputStream()
+
+        launch(coroutineContext) {
+            channel.cancel()
+        }
+
+        yield()
+
+        try {
+            output.use {
+                output.write(1)
+            }
+            fail("write() should fail")
+        } catch (expected: CancellationException) {
+        }
+    }
 }