Dispatchers default behaviour is changed to schedule new coroutine for execution later.

Job.getInactiveCancellationException is introduced.
yield is cancellable while it is waiting to executed.
core tutorial is expanded.

Author: elizarov

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

@@ -1,6 +1,8 @@
 package kotlinx.coroutines.experimental
 
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater
 import kotlin.coroutines.Continuation
+import kotlin.coroutines.ContinuationInterceptor
 import kotlin.coroutines.intrinsics.SUSPENDED_MARKER
 import kotlin.coroutines.intrinsics.suspendCoroutineOrReturn
 import kotlin.coroutines.suspendCoroutine
@@ -37,25 +39,39 @@ public inline suspend fun <T> suspendCancellableCoroutine(crossinline block: (Ca
 internal fun getParentJobOrAbort(cont: Continuation<*>): Job? {
     val job = cont.context[Job]
     // fast path when parent job is already complete (we don't even construct SafeCancellableContinuation object)
-    job?.isActive?.let { if (!it) throw CancellationException() }
+    if (job != null && !job.isActive) throw job.getInactiveCancellationException()
     return job
 }
 
 @PublishedApi
 internal class SafeCancellableContinuation<in T>(
         private val delegate: Continuation<T>,
-        parentJob: Job?
+        private val parentJob: Job?
 ) : AbstractCoroutine<T>(delegate.context), CancellableContinuation<T> {
     // only updated from the thread that invoked suspendCancellableCoroutine
-    private var suspendedThread: Thread? = Thread.currentThread()
+
+    @Volatile
+    private var decision = UNDECIDED
+
+    private companion object {
+        val DECISION: AtomicIntegerFieldUpdater<SafeCancellableContinuation<*>> =
+                AtomicIntegerFieldUpdater.newUpdater(SafeCancellableContinuation::class.java, "decision")
+
+        const val UNDECIDED = 0
+        const val SUSPENDED = 1
+        const val RESUMED = 2
+        const val YIELD = 3 // used by cancellable "yield"
+    }
 
     init { initParentJob(parentJob) }
 
     fun getResult(): Any? {
-        if (suspendedThread != null) {
-            suspendedThread = null
-            return SUSPENDED_MARKER
+        val decision = this.decision // volatile read
+        when (decision) {
+            UNDECIDED -> if (DECISION.compareAndSet(this, UNDECIDED, SUSPENDED)) return SUSPENDED_MARKER
+            YIELD -> return SUSPENDED_MARKER
         }
+        // otherwise, afterCompletion was already invoked, and the result is in the state
         val state = getState()
         if (state is CompletedExceptionally) throw state.exception
         return state
@@ -66,15 +82,21 @@ internal class SafeCancellableContinuation<in T>(
 
     @Suppress("UNCHECKED_CAST")
     override fun afterCompletion(state: Any?) {
-        if (suspendedThread === Thread.currentThread()) {
-            // cancelled during suspendCancellableCoroutine in its thread
-            suspendedThread = null
-        } else {
-            // cancelled later or in other thread
-            if (state is CompletedExceptionally)
-                delegate.resumeWithException(state.exception)
-            else
-                delegate.resume(state as T)
-        }
+        val decision = this.decision // volatile read
+        if (decision == UNDECIDED && DECISION.compareAndSet(this, UNDECIDED, RESUMED)) return // will get result in getResult
+        // otherwise, getResult has already commenced, i.e. it was resumed later or in other thread
+        if (state is CompletedExceptionally)
+            delegate.resumeWithException(state.exception)
+        else if (decision == YIELD && delegate is DispatchedContinuation)
+            delegate.resumeYield(parentJob, state as T)
+        else
+            delegate.resume(state as T)
+    }
+
+    // can only be invoked in the same thread as getResult (see "yield"), afterCompletion may be concurrent
+    fun resumeYield(value: T) {
+        if ((context[ContinuationInterceptor] as? CoroutineDispatcher)?.isDispatchNeeded(context) == true)
+            DECISION.compareAndSet(this, UNDECIDED, YIELD) // try mark as needing dispatch
+        resume(value)
     }
 }

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

@@ -1,6 +1,6 @@
 package kotlinx.coroutines.experimental
 
-import java.util.concurrent.Executor
+import java.util.concurrent.ExecutorService
 import java.util.concurrent.Executors
 import java.util.concurrent.atomic.AtomicInteger
 import kotlin.coroutines.CoroutineContext
@@ -12,21 +12,21 @@ import kotlin.coroutines.CoroutineContext
  * When available, it wraps `ForkJoinPool.commonPool` and provides a similar shared pool where not.
  */
 object CommonPool : CoroutineDispatcher() {
-    private val pool: Executor = findPool()
+    private val pool: ExecutorService = findPool()
 
     private inline fun <T> Try(block: () -> T) = try { block() } catch (e: Throwable) { null }
 
-    private fun findPool(): Executor {
+    private fun findPool(): ExecutorService {
         val fjpClass = Try { Class.forName("java.util.concurrent.ForkJoinPool") }
             ?: return createPlainPool()
-        Try { fjpClass.getMethod("commonPool")?.invoke(null) as? Executor }
+        Try { fjpClass.getMethod("commonPool")?.invoke(null) as? ExecutorService }
             ?. let { return it }
-        Try { fjpClass.getConstructor(Int::class.java).newInstance(defaultParallelism()) as? Executor }
+        Try { fjpClass.getConstructor(Int::class.java).newInstance(defaultParallelism()) as? ExecutorService }
             ?. let { return it }
         return createPlainPool()
     }
 
-    private fun createPlainPool(): Executor {
+    private fun createPlainPool(): ExecutorService {
         val threadId = AtomicInteger()
         return Executors.newFixedThreadPool(defaultParallelism()) {
             Thread(it, "CommonPool-worker-${threadId.incrementAndGet()}").apply { isDaemon = true }
@@ -35,6 +35,5 @@ object CommonPool : CoroutineDispatcher() {
 
     private fun defaultParallelism() = (Runtime.getRuntime().availableProcessors() - 1).coerceAtLeast(1)
 
-    override fun isDispatchNeeded(context: CoroutineContext): Boolean = true
     override fun dispatch(context: CoroutineContext, block: Runnable) = pool.execute(block)
 }

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

@@ -25,8 +25,9 @@ public abstract class CoroutineDispatcher :
         AbstractCoroutineContextElement(ContinuationInterceptor), ContinuationInterceptor {
     /**
      * Return `true` if execution shall be dispatched onto another thread.
+     * The default behaviour for most dispatchers is to return `true`.
      */
-    public abstract fun isDispatchNeeded(context: CoroutineContext): Boolean
+    public open fun isDispatchNeeded(context: CoroutineContext): Boolean = true
 
     /**
      * Dispatches execution of a runnable [block] onto another thread in the given [context].
@@ -37,7 +38,7 @@ public abstract class CoroutineDispatcher :
             DispatchedContinuation<T>(this, continuation)
 }
 
-private class DispatchedContinuation<T>(
+internal class DispatchedContinuation<T>(
         val dispatcher: CoroutineDispatcher,
         val continuation: Continuation<T>
 ): Continuation<T> by continuation {
@@ -68,4 +69,22 @@ private class DispatchedContinuation<T>(
                 continuation.resumeWithException(exception)
             }
     }
+
+    // used by "yield" implementation
+    fun resumeYield(job: Job?, value: T) {
+        val context = continuation.context
+        if (dispatcher.isDispatchNeeded(context))
+            dispatcher.dispatch(context, Runnable {
+                withCoroutineContext(context) {
+                    if (job?.isActive == false)
+                        continuation.resumeWithException(job.getInactiveCancellationException())
+                    else
+                        continuation.resume(value)
+                }
+            })
+        else
+            withCoroutineContext(context) {
+                continuation.resume(value)
+            }
+    }
 }

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

@@ -3,7 +3,6 @@ package kotlinx.coroutines.experimental
 import kotlinx.coroutines.experimental.internal.LockFreeLinkedListHead
 import kotlinx.coroutines.experimental.internal.LockFreeLinkedListNode
 import java.util.concurrent.locks.LockSupport
-import kotlin.coroutines.Continuation
 import kotlin.coroutines.CoroutineContext
 
 /**
@@ -41,7 +40,7 @@ public interface EventLoop {
 
 internal class EventLoopImpl(
     val thread: Thread
-) : CoroutineDispatcher(), EventLoop, Yield {
+) : CoroutineDispatcher(), EventLoop {
     val queue = LockFreeLinkedListHead()
     var parentJob: Job? = null
 
@@ -50,26 +49,18 @@ internal class EventLoopImpl(
         this.parentJob = coroutine
     }
 
-    override fun isDispatchNeeded(context: CoroutineContext): Boolean = Thread.currentThread() != thread
-
     override fun dispatch(context: CoroutineContext, block: Runnable) {
         schedule(Dispatch(block))
     }
 
-    override fun scheduleResume(continuation: CancellableContinuation<Unit>) {
-        val node = Resume(continuation)
-        if (schedule(node))
-            continuation.removeOnCompletion(node)
-    }
-
     fun schedule(node: Node): Boolean {
         val added = if (parentJob == null) {
             queue.addLast(node)
             true
         } else
             queue.addLastIf(node) { parentJob!!.isActive }
         if (added) {
-            if (Thread.currentThread() != thread)
+            if (Thread.currentThread() !== thread)
                 LockSupport.unpark(thread)
         } else {
             node.run()
@@ -78,7 +69,7 @@ internal class EventLoopImpl(
     }
 
     override fun processNextEvent(): Boolean {
-        if (Thread.currentThread() != thread) return false
+        if (Thread.currentThread() !== thread) return false
         (queue.removeFirstOrNull() as? Runnable)?.apply {
             run()
             return true
@@ -89,9 +80,5 @@ internal class EventLoopImpl(
     abstract class Node : LockFreeLinkedListNode(), Runnable
 
     class Dispatch(block: Runnable) : Node(), Runnable by block
-
-    class Resume(val cont: Continuation<Unit>) : Node() {
-        override fun run() = cont.resume(Unit)
-    }
 }
 

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

@@ -13,7 +13,6 @@ public fun Executor.toCoroutineDispatcher(): CoroutineDispatcher =
     ExecutorCoroutineDispatcher(this)
 
 internal open class ExecutorCoroutineDispatcher(val executor: Executor) : CoroutineDispatcher(), Delay {
-    override fun isDispatchNeeded(context: CoroutineContext): Boolean = true
     override fun dispatch(context: CoroutineContext, block: Runnable) = executor.execute(block)
 
     override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {

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

@@ -39,6 +39,13 @@ public interface Job : CoroutineContext.Element {
      */
     public val isActive: Boolean
 
+    /**
+     * Returns [CancellationException] that [cancellable][suspendCancellableCoroutine] suspending functions throw when
+     * trying to suspend in the context of this job. This function throws [IllegalAccessException] when invoked
+     * for an [active][isActive] job.
+     */
+    fun getInactiveCancellationException(): CancellationException
+
     /**
      * Registers completion handler. The action depends on the state of this job.
      * When job is cancelled with [cancel], then the handler is immediately invoked
@@ -230,6 +237,15 @@ internal open class JobSupport : AbstractCoroutineContextElement(Job), Job {
 
     final override val isActive: Boolean get() = state is Active
 
+    override fun getInactiveCancellationException(): CancellationException {
+        val state = getState()
+        return when (state) {
+            is Active -> throw IllegalStateException("Job is still active")
+            is CompletedExceptionally -> state.cancellationException
+            else -> CancellationException("Job has completed with result")
+        }
+    }
+
     final override fun onCompletion(handler: CompletionHandler): Job.Registration {
         var nodeCache: JobNode? = null
         while (true) { // lock-free loop on state
@@ -332,6 +348,17 @@ internal open class JobSupport : AbstractCoroutineContextElement(Job), Job {
     internal abstract class CompletedExceptionally {
         abstract val cancelReason: Throwable // original reason or fresh CancellationException
         abstract val exception: Throwable // the exception to be thrown in continuation
+
+        // convert cancelReason to CancellationException on first need
+        @Volatile
+        private var _cancellationException: CancellationException? = null
+
+        val cancellationException: CancellationException get() =
+            _cancellationException ?: // atomic read volatile var or else build new
+                (cancelReason as? CancellationException ?:
+                        CancellationException(cancelReason.message).apply { initCause(cancelReason) })
+                        .also { _cancellationException = it }
+
     }
 
     /**
@@ -343,24 +370,16 @@ internal open class JobSupport : AbstractCoroutineContextElement(Job), Job {
 
         override val cancelReason: Throwable get() =
             _cancelReason ?: // atomic read volatile var or else create new
-                CancellationException().also { _cancelReason = it }
+                CancellationException("Job was cancelled without specified reason").also { _cancelReason = it }
 
-        @Volatile
-        private var _exception: Throwable? = null // convert reason to CancellationException on first need
-
-        override val exception: Throwable get() =
-            _exception ?: // atomic read volatile var or else build new
-                (cancelReason as? CancellationException ?:
-                        CancellationException(cancelReason.message).apply { initCause(cancelReason) })
-                            .also { _exception = it }
+        override val exception: Throwable get() = cancellationException
     }
 
     /**
      * Represents a [state][getState] of a failed job.
      */
     internal class Failed(override val exception: Throwable) : CompletedExceptionally() {
-        override val cancelReason: Throwable
-            get() = exception
+        override val cancelReason: Throwable get() = exception
     }
 }
 

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

@@ -5,11 +5,11 @@ import java.util.concurrent.ScheduledThreadPoolExecutor
 import java.util.concurrent.TimeUnit
 import kotlin.coroutines.startCoroutine
 
-val KEEP_ALIVE = java.lang.Long.getLong("kotlinx.coroutines.ScheduledExecutor.keepAlive", 50L)
+val KEEP_ALIVE = java.lang.Long.getLong("kotlinx.coroutines.ScheduledExecutor.keepAlive", 1000L)
 
 internal val scheduledExecutor by lazy<ScheduledExecutorService> {
     ScheduledThreadPoolExecutor(1) { r ->
-        Thread(r, "kotlinx.coroutines.ScheduledExecutor")
+        Thread(r, "kotlinx.coroutines.ScheduledExecutor").apply { isDaemon = true }
     }.apply {
         setKeepAliveTime(KEEP_ALIVE, TimeUnit.MILLISECONDS)
         allowCoreThreadTimeOut(true)

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

@@ -4,7 +4,6 @@ import java.util.concurrent.Executors
 import java.util.concurrent.ScheduledExecutorService
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicInteger
-import kotlin.concurrent.thread
 import kotlin.coroutines.CoroutineContext
 
 /**
@@ -30,34 +29,22 @@ fun newFixedThreadPoolContext(nThreads: Int, name: String, parent: Job? = null):
     return job + ThreadPoolDispatcher(nThreads, name, job)
 }
 
-private val thisThreadContext = ThreadLocal<ThreadPoolDispatcher>()
-
 private class ThreadPoolDispatcher(
         nThreads: Int,
         name: String,
         val job: Job
-) : CoroutineDispatcher(), Yield, Delay {
+) : CoroutineDispatcher(), Delay {
     val threadNo = AtomicInteger()
     val executor: ScheduledExecutorService = Executors.newScheduledThreadPool(nThreads) { target ->
-        thread(start = false, isDaemon = true,
-                name = if (nThreads == 1) name else name + "-" + threadNo.incrementAndGet()) {
-            thisThreadContext.set(this@ThreadPoolDispatcher)
-            target.run()
-        }
+        Thread(target, if (nThreads == 1) name else name + "-" + threadNo.incrementAndGet()).apply { isDaemon = true }
     }
 
     init {
         job.onCompletion { executor.shutdown() }
     }
 
-    override fun isDispatchNeeded(context: CoroutineContext): Boolean = thisThreadContext.get() != this
-
     override fun dispatch(context: CoroutineContext, block: Runnable) = executor.execute(block)
 
-    override fun scheduleResume(continuation: CancellableContinuation<Unit>) {
-        executor.scheduleResume(continuation)
-    }
-
     override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
         executor.scheduleResumeAfterDelay(time, unit, continuation)
     }