BlockingPool implementation.

Author: fvasco

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

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kotlinx.coroutines.experimental
+
+import java.util.concurrent.Executor
+import java.util.concurrent.Executors
+import kotlin.coroutines.experimental.CoroutineContext
+
+/**
+ * This is a common pool of shared threads for blocking tasks.
+ *
+ * It uses a *unbounded* cached thread pool.
+ */
+object BlockingPool : CoroutineDispatcher(), Executor {
+
+    private val threadGroup = ThreadGroup("kotlinx.coroutines.BlockingPool")
+
+    private val executor = Executors.newCachedThreadPool({
+        Thread(threadGroup, it, "kotlinx.coroutines.BlockingPool-${System.identityHashCode(it).toString(16)}").apply {
+            isDaemon = true
+        }
+    })
+
+    override fun execute(command: Runnable) = executor.execute(command)
+
+    override fun dispatch(context: CoroutineContext, block: Runnable) {
+        execute(block)
+    }
+
+    internal fun isDispatchNeeded(): Boolean = threadGroup !== Thread.currentThread().threadGroup
+
+    override fun isDispatchNeeded(context: CoroutineContext): Boolean = isDispatchNeeded()
+}

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

@@ -24,6 +24,39 @@ import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
 
 // --------------- basic coroutine builders ---------------
 
+/**
+ * Execute the given [block] in an unbounded executor.
+ * If the current thread is a blocking one then the [block] is executed immediately,
+ * otherwise the coroutine is suspended and the [block] is scheduled for execution on a different thread.
+ */
+public suspend fun <T> blocking(block: () -> T): T =
+        if (!BlockingPool.isDispatchNeeded())
+            // execute block undispatched
+            block()
+        else
+            suspendCancellableCoroutine { continuation ->
+                val runnable = object : Runnable {
+                    @JvmField
+                    @Volatile
+                    var uncompletedContinuation: CancellableContinuation<T>? = continuation
+
+                    override fun run() {
+                        val res =
+                                try {
+                                    block()
+                                } catch (exception: Throwable) {
+                                    uncompletedContinuation?.resumeWithException(exception)
+                                    return
+                                }
+                        uncompletedContinuation?.resume(res)
+                    }
+                }
+
+                // execute block in blocking pool
+                BlockingPool.execute(runnable)
+                continuation.invokeOnCompletion { runnable.uncompletedContinuation = null }
+            }
+
 /**
  * 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].

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kotlinx.coroutines.experimental
+
+import guide.test.ignoreLostThreads
+import org.junit.Test
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+class BlockingTest : TestBase() {
+    @Test
+    fun testBlockingDelay() {
+        ignoreLostThreads("kotlinx.coroutines.BlockingPool")
+        runBlocking {
+            val taskCount = Runtime.getRuntime().availableProcessors() * 3
+            val countDown = CountDownLatch(taskCount)
+            repeat(taskCount) {
+                async {
+                    assert(BlockingPool.isDispatchNeeded())
+                    blocking {
+                        assert(!BlockingPool.isDispatchNeeded())
+                        Thread.sleep(50L)
+                        countDown.countDown()
+                    }
+                }
+            }
+            countDown.await(80L, TimeUnit.MILLISECONDS)
+        }
+    }
+}