Support for blocking calls inside coroutines (Kotlin#79)

Author: Francesco Vasco

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

@@ -16,6 +16,7 @@
 
 package kotlinx.coroutines.experimental
 
+import java.util.concurrent.Executor
 import java.util.concurrent.locks.LockSupport
 import kotlin.coroutines.experimental.*
 import kotlin.coroutines.experimental.intrinsics.COROUTINE_SUSPENDED
@@ -24,6 +25,24 @@ 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 = defaultBlockingExecutor, 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/DefaultExecutor.kt

@@ -16,22 +16,41 @@
 
 package kotlinx.coroutines.experimental
 
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.LinkedBlockingQueue
 import java.util.concurrent.ScheduledExecutorService
 import java.util.concurrent.ScheduledThreadPoolExecutor
+import java.util.concurrent.ThreadPoolExecutor
 import java.util.concurrent.TimeUnit
 
 private const val DEFAULT_KEEP_ALIVE = 1000L
+private const val DEFAULT_BLOCKING_KEEP_ALIVE = 60_000L
+private const val DEFAULT_BLOCKING_POOL_SIZE = 256
 
 private val KEEP_ALIVE =
     try { java.lang.Long.getLong("kotlinx.coroutines.DefaultExecutor.keepAlive", DEFAULT_KEEP_ALIVE) }
     catch (e: SecurityException) { DEFAULT_KEEP_ALIVE }
 
+private val BLOCKING_KEEP_ALIVE =
+    try { java.lang.Long.getLong("kotlinx.coroutines.DefaultBlockingExecutor.keepAlive", DEFAULT_BLOCKING_KEEP_ALIVE) }
+    catch (e: SecurityException) { DEFAULT_BLOCKING_KEEP_ALIVE }
+
+private val BLOCKING_POOL_SIZE =
+    try { java.lang.Integer.getInteger("kotlinx.coroutines.DefaultBlockingExecutor.maximumPoolSize", DEFAULT_BLOCKING_POOL_SIZE) }
+    catch (e: SecurityException) { DEFAULT_BLOCKING_POOL_SIZE }
+
 @Volatile
 private var _executor: ScheduledExecutorService? = null
 
+@Volatile
+private var _blockingExecutor: ExecutorService? = null
+
 internal val defaultExecutor: ScheduledExecutorService
     get() = _executor ?: getOrCreateExecutorSync()
 
+internal val defaultBlockingExecutor: ExecutorService
+    get() = _blockingExecutor ?: getOrCreateBlockingExecutorSync()
+
 @Synchronized
 private fun getOrCreateExecutorSync(): ScheduledExecutorService =
     _executor ?: ScheduledThreadPoolExecutor(1) { r ->
@@ -48,6 +67,17 @@ private fun getOrCreateExecutorSync(): ScheduledExecutorService =
         _executor = this
     }
 
+@Synchronized
+private fun getOrCreateBlockingExecutorSync(): ExecutorService =
+    _blockingExecutor ?: ThreadPoolExecutor(1, BLOCKING_POOL_SIZE,
+        BLOCKING_KEEP_ALIVE, TimeUnit.MILLISECONDS,
+        LinkedBlockingQueue<Runnable>()) { r ->
+        Thread(r, "kotlinx.coroutines.DefaultBlockingExecutor").apply { isDaemon = true }
+    }.apply {
+        allowCoreThreadTimeOut(true)
+        _blockingExecutor = this
+    }
+
 // used for tests
 @Synchronized
 internal fun shutdownDefaultExecutor(timeout: Long) {

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

@@ -19,6 +19,7 @@ package kotlinx.coroutines.experimental
 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
 
 /**
@@ -160,6 +161,21 @@ 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.
+ * [start] policy is same as [async] method, but [CoroutineStart.UNDISPATCHED] is an invalid option
+ *
+ * @param executor the executor for blocking code
+ * @param start start option, [CoroutineStart.UNDISPATCHED] is invalid
+ * @param block the blocking code
+ */
+public fun <T> blockingAsync(executor: Executor = defaultBlockingExecutor,
+                             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/test/kotlin/kotlinx/coroutines/experimental/AsyncTest.kt

@@ -144,6 +144,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()
@@ -154,4 +171,4 @@ class AsyncTest : TestBase() {
         assertTrue(d.await() === bad)
         finish(2)
     }
-}
+}