Support optional thread interrupt in newSingleThreadContext as its friends

[elizarov]

See discussion here: https://discuss.kotlinlang.org/t/calling-blocking-code-in-coroutines/2368

[Tolriq]

@elizarov sorry to post here but I'm sure this could help a lot of other people.

I really struggle to understand how to properly cancel blocking IO calls since normal usage does not interrupt thread.

Goal would be convert some functions in a module to coroutines with blocking IO and some computation.

I get the isActive/yield and ways to cancel the computations, but not how to push the cancel to the blocking IO call.

If I understand correctly how to properly do the coroutines stuff it should be something like this

suspend fun x(): y {
   ....
   val result = withContext(IO) {
   }
   do CPU computation checking for isActive()/yield
}

But from all that I read, cancelling the coroutine that call x() would still have to wait for the result to finish it's IO blocking call that only support thread interrupt)

Is there any usable documentation / sample that fully cover this need (for Android so without Guava if possible)

(Plan is not to use IO but an existing executor with asCoroutineDispatcher())

[khovanskiy]

I meet the issue like in the references, so the following workaround was implemented:

val externalThreadPool = Executors.newCachedThreadPool()
suspend fun <T> withTimeoutOrInterrupt(timeMillis: Long, block: () -> T) {
    withTimeout(timeMillis) {
        suspendCancellableCoroutine<T> { cont ->
            val future = externalThreadPool.submit {
                try {
                    val result = block()
                    cont.resumeWith(Result.success(result))
                } catch (e: InterruptedException) {
                    cont.resumeWithException(CancellationException())
                } catch (e: Throwable) {
                    cont.resumeWithException(e);
                }
            }
            cont.invokeOnCancellation {
                future.cancel(true)
            }
        }
    }
}

It provides the similar behaviour like usual withTimeout, but additionally it supports running code with blocking.
Note: It should be called only when you know, that the inner code use blocking and can correctly process throwed InterruptedException. In most cases withTimeout function is prefered.

[elizarov]

@khovanskiy This "works", but it is extremely dangerous code to use in a serious high-load production. Here's the implementation of that future.cancel method from Open JDK for your reference:
https://github.com/openjdk/jdk/blob/67a89143dde6e545adbfc3c79bb89d954307f8bc/src/java.base/share/classes/java/util/concurrent/FutureTask.java#L164-L182

Ask you can see it does a classic "check & act" concurrent programming mistake: It interrupts a thread that was running this task at the time it had checked it, so under load it may, in fact, interrupt some other, unrelated piece of code.

[elizarov]

UPDATE: I stand corrected. False alarm. This particular usage of Executors.newCachedThreadPool() with future.cancel(true) seems to be fine. All bases are covered. See here: https://twitter.com/relizarov/status/1184460504238100480

[jxdabc]

@elizarov

I think this ability is of great need on Android.

According to

• coroutine doc: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-main-scope.html
• Android guide: https://developer.android.com/kotlin/coroutines

The practical usage on Android would be something like:

class MyAndroidActivity {
    private val scope = MainScope()

    override fun onResume {
         scope.launch {
             var info: Info
             withContext(Dispatchers.IO) {
                 // #1 may block
                info = loadFromInternet()
             }
             showOnUI(info)
         }
    }

    override fun onDestroy() {
        super.onDestroy()
        scope.cancel()
    }
}

Line #1 may block for a relatively long time. If a lot of activities like this are entered and exited in a relatively short time, eg., someone is looking for interesting news and entering exiting news detail pages quickly without long stay, or someone is doing monkey test. The Dispatcher will be filled with canceled but blocked coroutines, so that new info loading of new activities are starving and unable to run.

[jxdabc]

@elizarov

To implement this, I think, we could

1. Make it optional and default off -- explicitly use only.
2. Watch the job for cancellation and interrupt the executing thread when, and only when, continuations of a coroutine is running.
3. Transform InterruptedException to CancellationException at the end of the coroutine so that everything else goes as before.

• update/restore of ThreadContextElement is perfect place to implement 2
• To implement 3, transform before AbstractCoroutine#makeCompletingOnce requiring a little modification
• Implementing 1 with a context element is easy

[jxdabc]

I made a PR to implement this #1922
The added context element is something like:

/**
 * 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)

[jxdabc]

PR (non-intrusive implementation of this feature): #1934

As mentioned above. This feature is of great use at least on Android as I could see. There is still another non-intrusive implementation of this feature: Add a function that makes a blocking code block cancellable (become a cancellation point of the coroutine) by interrupting the blocking code block and throwing CancellationException on coroutine cancellation.

The added function is like:

/**
 * Makes a blocking code block cancellable (become a cancellation point of the coroutine).
 *
 * The blocking code block will be interrupted and this function will throw [CancellationException]
 * if the coroutine is cancelled.
 *
 * Example:
 * ```
 * GlobalScope.launch(Dispatchers.IO) {
 *     async {
 *         // This function will throw [CancellationException].
 *         interruptible {
 *             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()
 *     }
 * }
 * ```
 */
public suspend fun <T> interruptible(block: () -> T): T

[elizarov]

👍 interruptible { ... } block seems to be lean an isolated way to implement it. Let's move it to a separate issue, though -> #1947 . I'm closing this one.

[AndroidDeveloperLB]

Is it possible for Kotlin Coroutines to handle thread interruption?
If I call some function that has "sleep", for example, is it possible to cancel it so that it will also call "interrupt" ?

[qwwdfsad]

@AndroidDeveloperLB yes, you could use runInterruptible builder for this purpose

[AndroidDeveloperLB]

@qwwdfsad This almost works.
Almost because then it forces you to cancel by interruption, always.
Via AsyncTask, you always had a choice of whether cancelling normally or by interruptions.
Here it seems you have to set it up and stay with one of them, forever.

[elizarov]

What do you mean by "forces you to cancel by interruption"? Can you elaborate, please?

[AndroidDeveloperLB]

@elizarov Either you use runInterruptible or you don't.
If you use it, cancel will interrupt.
If you don't, cancel will not interrupt.
The creation of the instance forces cancel to work in one way or the other. You have no choice after it's created.

On AsyncTask, you have a choice for every instance. cancel(true) will interrupt, and cancel(false) will not.
The creation of the instance doesn't force you to choose how it will be canceled.
You can choose to cancel with interruption in some case, and without in another.