Support for blocking calls inside coroutines (#79)

Author: Francesco Vasco

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt

@@ -16,13 +16,32 @@
 
 package kotlinx.coroutines.experimental
 
+import java.util.concurrent.Executor
 import java.util.concurrent.locks.LockSupport
 import kotlin.coroutines.experimental.*
 import kotlin.coroutines.experimental.intrinsics.startCoroutineUninterceptedOrReturn
 import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
 
 // --------------- basic coroutine builders ---------------
 
+/**
+ * Suspend current coroutine and execute the [block] blocking code in [executor].
+ * The current coroutine is resumed after [block] execution.
+ *
+ * @param executor the executor for blocking code
+ * @param block the blocking code
+ */
+suspend fun <T> blocking(executor: Executor = IOExecutor, block: () -> T) =
+        suspendCoroutine<T> { cont ->
+            executor.execute {
+                try {
+                    cont.resume(block())
+                } catch (t: Throwable) {
+                    cont.resumeWithException(t)
+                }
+            }
+        }
+
 /**
  * Launches new coroutine without blocking current thread and returns a reference to the coroutine as a [Job].
  * The coroutine is cancelled when the resulting job is [cancelled][Job.cancel].

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Deferred.kt

@@ -20,8 +20,8 @@ import kotlinx.coroutines.experimental.intrinsics.startCoroutineUndispatched
 import kotlinx.coroutines.experimental.selects.SelectBuilder
 import kotlinx.coroutines.experimental.selects.SelectInstance
 import kotlinx.coroutines.experimental.selects.select
+import java.util.concurrent.Executor
 import kotlin.coroutines.experimental.CoroutineContext
-import kotlin.coroutines.experimental.startCoroutine
 
 /**
  * Deferred value is a non-blocking cancellable future.
@@ -137,6 +137,22 @@ public fun <T> async(
 public fun <T> async(context: CoroutineContext, start: Boolean, block: suspend CoroutineScope.() -> T): Deferred<T> =
     async(context, if (start) CoroutineStart.DEFAULT else CoroutineStart.LAZY, block)
 
+/**
+ * Execute the blocking code [block] in the [executor] without blocking current coroutine.
+ * [block] code is immediately scheduled for execution unless you set [start] as [CoroutineStart.LAZY], see [async] for more details.
+ *
+ * @param executor the executor for blocking code
+ * @param start start option, [CoroutineStart.UNDISPATCHED] is invalid
+ * @param block the blocking code
+ * @see async
+ */
+public fun <T> blockingAsync(executor: Executor = IOExecutor,
+                             start: CoroutineStart = CoroutineStart.DEFAULT,
+                             block: () -> T): Deferred<T> {
+    require(start != CoroutineStart.UNDISPATCHED) { "Start blocking code undispatched is not supported" }
+    return async(executor.asCoroutineDispatcher(), start) { block() }
+}
+
 /**
  * @suppress **Deprecated**: `defer` was renamed to `async`.
  */

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Executors.kt

@@ -18,10 +18,29 @@ package kotlinx.coroutines.experimental
 
 import java.util.concurrent.Executor
 import java.util.concurrent.ScheduledExecutorService
+import java.util.concurrent.SynchronousQueue
+import java.util.concurrent.ThreadFactory
+import java.util.concurrent.ThreadPoolExecutor
 import java.util.concurrent.TimeUnit
 import kotlin.coroutines.experimental.Continuation
 import kotlin.coroutines.experimental.CoroutineContext
 
+/**
+ * Create a shared Executor for blocking method invocation (ie: I/O, synchronized block, lock, etc...).
+ *
+ * You can tune the pool size and the thread lifetime setting the properties
+ * `kotlinx.coroutines.experimental.IOExecutor.maximumPoolSize` and
+ * `kotlinx.coroutines.experimental.IOExecutor.keepAliveMillis` respectively.
+ */
+public val IOExecutor: Executor = ThreadPoolExecutor(
+        0,
+        System.getProperty("kotlinx.coroutines.experimental.IOExecutor.maximumPoolSize").toIntOrNull()
+                ?: Runtime.getRuntime().availableProcessors() * 64,
+        System.getProperty("kotlinx.coroutines.experimental.IOExecutor.keepAliveMillis").toLongOrNull()
+                ?: 60_000L, TimeUnit.MILLISECONDS,
+        SynchronousQueue<Runnable>(),
+        ThreadFactory { r -> Thread(r, "IOExecutor-${Math.abs(System.identityHashCode(r)).toString(16)}") })
+
 /**
  * Converts an instance of [Executor] to an implementation of [CoroutineDispatcher].
  * @suppress **Deprecated**: Renamed to [asCoroutineDispatcher].

kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/AsyncTest.kt

@@ -110,6 +110,23 @@ class AsyncTest : TestBase() {
         override fun toString(): String = error("toString")
     }
 
+    @Test
+    fun blockingAsync() = runBlocking {
+        val d = blockingAsync { 42 }
+        check(d.await() == 42)
+    }
+
+    @Test
+    fun blockingAsyncLazy() = runBlocking {
+        val d = blockingAsync(start = CoroutineStart.LAZY) { 42 }
+        check(d.await() == 42)
+    }
+
+    @Test(expected = IllegalArgumentException::class)
+    fun blockingAsyncUndispatched() = runBlocking {
+        blockingAsync(start = CoroutineStart.UNDISPATCHED) { 42 }
+    }
+
     @Test
     fun testDeferBadClass() = runBlocking {
         val bad = BadClass()
@@ -120,4 +137,4 @@ class AsyncTest : TestBase() {
         assertTrue(d.await() === bad)
         finish(2)
     }
-}
+}