Implement optional thread interrupt on coroutine cancellation (#57) #1922
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
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
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) -> { | ||
|
There was a problem hiding this comment. If you swap initThread and initInvokeOnCancel then There was a problem hiding this comment. Seems a great idea, I will check it |
||
| 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.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Perhaps
when (s) {?