forked from Kotlin/kotlinx.coroutines
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implement optional thread interrupt on coroutine cancellation (Kotlin#57
) See issue Kotlin#57 for details Signed-off-by: Trol <jiaoxiaodong@xiaomi.com>
- Loading branch information
Trol
committed
Apr 19, 2020
1 parent
5eaf83c
commit 3565690
Showing
6 changed files
with
424 additions
and
2 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
55 changes: 55 additions & 0 deletions
55
kotlinx-coroutines-core/common/src/CoroutineInterruptController.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| /* | ||
| * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license. | ||
| */ | ||
|
|
||
| package kotlinx.coroutines | ||
|
|
||
| import kotlin.coroutines.AbstractCoroutineContextElement | ||
| import kotlin.coroutines.CoroutineContext | ||
|
|
||
| /** | ||
| * This [CoroutineContext] element makes a coroutine interruptible. | ||
| * | ||
| * With this element, the thread executing the coroutine is interrupted when the coroutine is canceled, making | ||
| * blocking procedures stop. Exceptions that indicate an interrupted procedure, eg., InterruptedException on JVM | ||
| * are transformed into [CancellationException] at the end of the coroutine. Thus, everything else goes as if this | ||
| * element is not present. In particular, the parent coroutine won't be canceled by those exceptions. | ||
| * | ||
| * This is an abstract element and will be implemented by each individual platform (or won't be implemented). | ||
| * The JVM implementation is named CoroutineInterruptible. | ||
| * | ||
| * Example: | ||
| * ``` | ||
| * GlobalScope.launch(Dispatchers.IO + CoroutineInterruptible) { | ||
| * async { | ||
| * // This block will throw [CancellationException] instead of an exception indicating | ||
| * // interruption, such as InterruptedException on JVM. | ||
| * withContext(CoroutineName) { | ||
| * doSomethingUseful() | ||
| * | ||
| * // This blocking procedure will be interrupted when this coroutine is canceled | ||
| * // by Exception thrown by the below async block. | ||
| * doSomethingElseUsefulInterruptible() | ||
| * } | ||
| * } | ||
| * | ||
| * async { | ||
| * delay(500L) | ||
| * throw Exception() | ||
| * } | ||
| * } | ||
| * ``` | ||
| */ | ||
| abstract class CoroutineInterruptController : AbstractCoroutineContextElement(Key) { | ||
| /** | ||
| * Key for [CoroutineInterruptController] instance in the coroutine context. | ||
| */ | ||
| @InternalCoroutinesApi | ||
| companion object Key : CoroutineContext.Key<CoroutineInterruptController> | ||
|
|
||
| /** | ||
| * Update the complete state of a coroutine, mainly for exception transformation. | ||
| */ | ||
| @InternalCoroutinesApi | ||
| abstract fun updateCoroutineCompleteState(completeState: Any?): Any? | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
152 changes: 152 additions & 0 deletions
152
kotlinx-coroutines-core/jvm/src/CoroutineInterruptible.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,152 @@ | ||
| /* | ||
| * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license. | ||
| */ | ||
|
|
||
| package kotlinx.coroutines | ||
|
|
||
| import kotlinx.atomicfu.AtomicRef | ||
| import kotlinx.atomicfu.atomic | ||
| import kotlinx.atomicfu.loop | ||
| import kotlin.coroutines.CoroutineContext | ||
|
|
||
| /** | ||
| * This is the [CoroutineInterruptController] implementation on JVM. See [CoroutineInterruptController] for detailed | ||
| * description and examples. | ||
| */ | ||
| object CoroutineInterruptible : | ||
| CoroutineInterruptController(), ThreadContextElement<CoroutineInterruptible.ThreadState?> { | ||
|
|
||
| /** | ||
| * Update the complete state of a coroutine on JVM. | ||
| * Transforms [InterruptedException] into [CancellationException] for coroutines with this context element. | ||
| */ | ||
| @InternalCoroutinesApi | ||
| override fun updateCoroutineCompleteState(completeState: Any?): Any? = | ||
| if (completeState is CompletedExceptionally && completeState.cause is InterruptedException) | ||
| CompletedExceptionally(CancellationException()) | ||
| else | ||
| completeState | ||
|
|
||
| /** | ||
| * Updates context of the current thread. | ||
| * This function is invoked before the coroutine in the specified [context] is resumed in the current thread. | ||
| * Prepares interruption for this execution, watching the [Job] for cancellation and interrupt this executing | ||
| * thread on cancellation. | ||
| */ | ||
| @InternalCoroutinesApi | ||
| override fun updateThreadContext(context: CoroutineContext): ThreadState? { | ||
| // Fast path: no Job in this context | ||
| val job = context[Job] ?: return null | ||
| // Slow path | ||
| val threadState = ThreadState() | ||
| threadState.initInterrupt(job) | ||
| return threadState | ||
| } | ||
|
|
||
| /** | ||
| * Restores context of the current thread. | ||
| * This function is invoked after the coroutine in the specified [context] is suspended in the current thread. | ||
| * Stops watching the [Job] for cancellation and do clean-up work. | ||
| */ | ||
| @InternalCoroutinesApi | ||
| override fun restoreThreadContext(context: CoroutineContext, oldState: ThreadState?) { | ||
| // Fast path: no Job in this context | ||
| val threadState = oldState ?: return | ||
| // Slow path | ||
| threadState.clearInterrupt() | ||
| } | ||
|
|
||
| /** | ||
| * Holds the state of executions for interruption. | ||
| */ | ||
| @InternalCoroutinesApi | ||
| class ThreadState { | ||
| fun initInterrupt(job: Job) { | ||
| initInvokeOnCancel(job) | ||
| initThread() | ||
| } | ||
|
|
||
| fun clearInterrupt() { | ||
| state.loop { s -> | ||
| when { | ||
| s is Working -> { | ||
| if (state.compareAndSet(s, Finish)) { | ||
| s.cancelHandle?.let { it.dispose() } // no more watching | ||
| return | ||
| } | ||
| } | ||
| s === Interrupting -> Thread.yield() // eases the thread | ||
| s === Interrupted -> { Thread.interrupted(); return } // no interrupt leak | ||
| s === Init || s === Finish -> throw IllegalStateException("impossible state") | ||
| else -> throw IllegalStateException("unknown state") | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private fun initInvokeOnCancel(job: Job) { | ||
| // watches the job for cancellation | ||
| val cancelHandle = | ||
| job.invokeOnCompletion(onCancelling = true, invokeImmediately = true, handler = CancelHandler()) | ||
| // remembers the cancel handle or drops it | ||
| state.loop { s -> | ||
| when { | ||
| s === Init -> if (state.compareAndSet(s, Working(null, cancelHandle))) return | ||
| s is Working -> if (state.compareAndSet(s, Working(s.thread, cancelHandle))) return | ||
| s === Finish -> { cancelHandle.dispose(); return } // no more watching needed | ||
| s === Interrupting || s === Interrupted -> return | ||
| else -> throw IllegalStateException("unknown state") | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private fun initThread() { | ||
| val thread = Thread.currentThread() | ||
| state.loop { s -> | ||
| when { | ||
| s === Init -> if (state.compareAndSet(s, Working(thread, null))) return | ||
| s is Working -> if (state.compareAndSet(s, Working(thread, s.cancelHandle))) return | ||
| s === Interrupted -> { thread.interrupt(); return } // interrupted before the thread is set | ||
| s === Finish || s === Interrupting -> throw IllegalStateException("impossible state") | ||
| else -> throw IllegalStateException("unknown state") | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private inner class CancelHandler : CompletionHandler { | ||
| override fun invoke(cause: Throwable?) { | ||
| state.loop { s -> | ||
| when { | ||
| s === Init || (s is Working && s.thread === null) -> { | ||
| if (state.compareAndSet(s, Interrupted)) | ||
| return | ||
| } | ||
| s is Working -> { | ||
| if (state.compareAndSet(s, Interrupting)) { | ||
| s.thread!!.interrupt() | ||
| state.value = Interrupted | ||
| return | ||
| } | ||
| } | ||
| s === Finish -> return | ||
| s === Interrupting || s === Interrupted -> return | ||
| else -> throw IllegalStateException("unknown state") | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private val state: AtomicRef<State> = atomic(Init) | ||
|
|
||
| private interface State | ||
| // initial state | ||
| private object Init : State | ||
| // cancellation watching is setup and/or the continuation is running | ||
| private data class Working(val thread: Thread?, val cancelHandle: DisposableHandle?) : State | ||
| // the continuation done running without interruption | ||
| private object Finish : State | ||
| // interrupting this thread | ||
| private object Interrupting: State | ||
| // done interrupting | ||
| private object Interrupted: State | ||
| } | ||
| } |
Oops, something went wrong.