No Job in newSingleThreadContext and newFixedThreadPoolContext anymore

* This resolves the common issue of using `run(ctx)` where ctx comes from
either `newSingleThreadContext` or `newFixedThreadPoolContext`
invocation. They both used to return a combination of dispatcher + job,
and this job was overriding the parent job, thus preventing propagation
of cancellation. Not anymore.
* ThreadPoolDispatcher class is now public and is the result type for
both functions. It has the `close` method to release the thread pool.

Fixes #149
Fixes #151

Author: elizarov

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

@@ -39,9 +39,15 @@ public fun Executor.asCoroutineDispatcher(): CoroutineDispatcher =
 
 private class ExecutorCoroutineDispatcher(override val executor: Executor) : ExecutorCoroutineDispatcherBase()
 
-internal abstract class ExecutorCoroutineDispatcherBase : CoroutineDispatcher(), Delay {
-    abstract val executor: Executor
-
+/**
+ * @suppress **This is unstable API and it is subject to change.**
+ */
+public abstract class ExecutorCoroutineDispatcherBase : CoroutineDispatcher(), Delay {
+    /**
+     * @suppress **This is unstable API and it is subject to change.**
+     */
+    internal abstract val executor: Executor
+    
     override fun dispatch(context: CoroutineContext, block: Runnable) =
         try { executor.execute(timeSource.trackTask(block)) }
         catch (e: RejectedExecutionException) {

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

@@ -16,54 +16,77 @@
 
 package kotlinx.coroutines.experimental
 
+import java.io.Closeable
 import java.util.concurrent.Executors
 import java.util.concurrent.ScheduledExecutorService
 import java.util.concurrent.atomic.AtomicInteger
 import kotlin.coroutines.experimental.CoroutineContext
 
 /**
  * Creates new coroutine execution context with the a single thread and built-in [yield] and [delay] support.
- * All continuations are dispatched immediately when invoked inside the thread of this context.
- * Resources of this pool (its thread) are reclaimed when job of this context is cancelled.
- * The specified [name] defines the name of the new thread.
- * An optional [parent] job may be specified upon creation.
+ * **NOTE: The resulting [ThreadPoolDispatcher] owns native resources (its thread).
+ * Resources are reclaimed by [ThreadPoolDispatcher.close].**
+ *
+ * @param name the base name of the created thread.
  */
+fun newSingleThreadContext(name: String): ThreadPoolDispatcher =
+    newFixedThreadPoolContext(1, name)
+
+/**
+ * @suppress **Deprecated**: Parent job is no longer supported.
+ */
+@Deprecated(message = "Parent job is no longer supported, `close` the resulting ThreadPoolDispatcher to release resources",
+    level = DeprecationLevel.WARNING, replaceWith = ReplaceWith("newSingleThreadContext(name)"))
 fun newSingleThreadContext(name: String, parent: Job? = null): CoroutineContext =
-    newFixedThreadPoolContext(1, name, parent)
+    newFixedThreadPoolContext(1, name)
 
 /**
  * Creates new coroutine execution context with the fixed-size thread-pool and built-in [yield] and [delay] support.
- * All continuations are dispatched immediately when invoked inside the threads of this context.
- * Resources of this pool (its threads) are reclaimed when job of this context is cancelled.
- * The specified [name] defines the names of the threads.
- * An optional [parent] job may be specified upon creation.
+ * **NOTE: The resulting [ThreadPoolDispatcher] owns native resources (its threads).
+ * Resources are reclaimed by [ThreadPoolDispatcher.close].**
+ *
+ * @param nThreads the number of threads.
+ * @param name the base name of the created threads.
  */
-fun newFixedThreadPoolContext(nThreads: Int, name: String, parent: Job? = null): CoroutineContext {
+fun newFixedThreadPoolContext(nThreads: Int, name: String): ThreadPoolDispatcher {
     require(nThreads >= 1) { "Expected at least one thread, but $nThreads specified" }
-    val job = Job(parent)
-    return job + ThreadPoolDispatcher(nThreads, name, job)
+    return ThreadPoolDispatcher(nThreads, name)
 }
 
+/**
+ * @suppress **Deprecated**: Parent job is no longer supported.
+ */
+@Deprecated(message = "Parent job is no longer supported, `close` the resulting ThreadPoolDispatcher to release resources",
+    level = DeprecationLevel.WARNING, replaceWith = ReplaceWith("newFixedThreadPoolContext(nThreads, name)"))
+fun newFixedThreadPoolContext(nThreads: Int, name: String, parent: Job? = null): CoroutineContext =
+    newFixedThreadPoolContext(nThreads, name)
+
 internal class PoolThread(
     @JvmField val dispatcher: ThreadPoolDispatcher, // for debugging & tests
     target: Runnable, name: String
 ) : Thread(target, name) {
     init { isDaemon = true }
 }
 
-internal class ThreadPoolDispatcher(
+/**
+ * Dispatches coroutine execution to a thread pool of a fixed size. Instances of this dispatcher are
+ * created with [newSingleThreadContext] and [newFixedThreadPoolContext].
+ */
+public class ThreadPoolDispatcher internal constructor(
     private val nThreads: Int,
-    private val name: String,
-    job: Job
-) : ExecutorCoroutineDispatcherBase() {
+    private val name: String
+) : ExecutorCoroutineDispatcherBase(), Closeable {
     private val threadNo = AtomicInteger()
 
-    override val executor: ScheduledExecutorService = Executors.newScheduledThreadPool(nThreads) { target ->
+    internal override val executor: ScheduledExecutorService = Executors.newScheduledThreadPool(nThreads) { target ->
         PoolThread(this, target, if (nThreads == 1) name else name + "-" + threadNo.incrementAndGet())
     }
 
-    init {
-        job.invokeOnCompletion { executor.shutdown() }
+    /**
+     * Closes this dispatcher -- shuts down all threads in this pool and releases resources.
+     */
+    public override fun close() {
+        executor.shutdown()
     }
 
     override fun toString(): String = "ThreadPoolDispatcher[$nThreads, $name]"

core/kotlinx-coroutines-core/src/test/kotlin/guide/example-context-04.kt

@@ -22,13 +22,15 @@ import kotlinx.coroutines.experimental.*
 fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
 
 fun main(args: Array<String>) {
-    val ctx1 = newSingleThreadContext("Ctx1")
-    val ctx2 = newSingleThreadContext("Ctx2")
-    runBlocking(ctx1) {
-        log("Started in ctx1")
-        run(ctx2) {
-            log("Working in ctx2")
+    newSingleThreadContext("Ctx1").use { ctx1 ->
+        newSingleThreadContext("Ctx2").use { ctx2 ->
+            runBlocking(ctx1) {
+                log("Started in ctx1")
+                run(ctx2) {
+                    log("Working in ctx2")
+                }
+                log("Back to ctx1")
+            }
         }
-        log("Back to ctx1")
     }
 }

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

@@ -31,7 +31,7 @@ class ExecutorsTest : TestBase() {
         runBlocking(context) {
             checkThreadName("TestThread")
         }
-        context[Job]!!.cancel()
+        context.close()
     }
 
     @Test
@@ -40,7 +40,7 @@ class ExecutorsTest : TestBase() {
         runBlocking(context) {
             checkThreadName("TestPool")
         }
-        context[Job]!!.cancel()
+        context.close()
     }
 
     @Test
@@ -63,7 +63,7 @@ class ExecutorsTest : TestBase() {
             }
             checkThreadName("Ctx1")
         }
-        ctx1[Job]!!.cancel()
-        ctx2[Job]!!.cancel()
+        ctx1.close()
+        ctx2.close()
     }
 }

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/BroadcastChannelMultiReceiveStressTest.kt

@@ -52,7 +52,7 @@ class BroadcastChannelMultiReceiveStressTest(
 
     @After
     fun tearDown() {
-        pool.cancel()
+        pool.close()
     }
 
     @Test

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/ConflatedChannelCloseStressTest.kt

@@ -35,7 +35,9 @@ class ConflatedChannelCloseStressTest : TestBase() {
     val pool = newFixedThreadPoolContext(nSenders + 2, "TestStressClose")
 
     @After
-    fun tearDown() { pool[Job]!!.cancel() }
+    fun tearDown() {
+        pool.close()
+    }
 
     @Test
     fun testStressClose() = runBlocking<Unit> {

coroutines-guide.md

@@ -815,13 +815,17 @@ It produces the following output (maybe in different order):
 
 <!--- TEST LINES_START_UNORDERED -->
 
-The default dispatcher that we've used in previous sections is representend by [DefaultDispather], which 
+The default dispatcher that we've used in previous sections is representend by [DefaultDispatcher], which 
 is equal to [CommonPool] in the current implementation. So, `launch { ... }` is the same 
 as `launch(DefaultDispather) { ... }`, which is the same as `launch(CommonPool) { ... }`. 
 
 The difference between parent [coroutineContext][CoroutineScope.coroutineContext] and
 [Unconfined] context will be shown later.
 
+Note, that [newSingleThreadContext] creates a new thread, which is a very expensive resource. 
+In a real application it must be either released, when no longer needed, using [close][ThreadPoolDispatcher.close] 
+function, or stored in a top-level variable and reused throughout the application.  
+
 ### Unconfined vs confined dispatcher
 
 The [Unconfined] coroutine dispatcher starts coroutine in the caller thread, but only until the
@@ -925,22 +929,24 @@ Run the following code with `-Dkotlinx.coroutines.debug` JVM option:
 fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
 
 fun main(args: Array<String>) {
-    val ctx1 = newSingleThreadContext("Ctx1")
-    val ctx2 = newSingleThreadContext("Ctx2")
-    runBlocking(ctx1) {
-        log("Started in ctx1")
-        run(ctx2) {
-            log("Working in ctx2")
+    newSingleThreadContext("Ctx1").use { ctx1 ->
+        newSingleThreadContext("Ctx2").use { ctx2 ->
+            runBlocking(ctx1) {
+                log("Started in ctx1")
+                run(ctx2) {
+                    log("Working in ctx2")
+                }
+                log("Back to ctx1")
+            }
         }
-        log("Back to ctx1")
     }
 }
 ```
 
 > You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-context-04.kt)
 
-It demonstrates two new techniques. One is using [runBlocking] with an explicitly specified context, and
-the second one is using [run] function to change a context of a coroutine while still staying in the 
+It demonstrates several new techniques. One is using [runBlocking] with an explicitly specified context, and
+the other one is using [run] function to change a context of a coroutine while still staying in the 
 same coroutine as you can see in the output below:
 
 ```text
@@ -951,6 +957,10 @@ same coroutine as you can see in the output below:
 
 <!--- TEST -->
 
+
+Note, that is example also uses `use` function from the Kotlin standard library to release threads that
+are created with [newSingleThreadContext] when they are no longer needed. 
+
 ### Job in the context
 
 The coroutine's [Job] is part of its context. The coroutine can retrieve it from its own context 
@@ -2325,9 +2335,12 @@ Channel was closed
 [Deferred.await]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-deferred/await.html
 [Job.start]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-job/start.html
 [CoroutineDispatcher]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-dispatcher/index.html
+[DefaultDispatcher]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-default-dispatcher.html
 [CommonPool]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-common-pool/index.html
 [CoroutineScope.coroutineContext]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-scope/coroutine-context.html
 [Unconfined]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-unconfined/index.html
+[newSingleThreadContext]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/new-single-thread-context.html
+[ThreadPoolDispatcher.close]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-thread-pool-dispatcher/close.html
 [newCoroutineContext]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/new-coroutine-context.html
 [CoroutineName]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-name/index.html
 [Job()]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-job.html