How to cancel with and without interruption using Coroutines on Android, including auto-cancellation according to lifecycle?

Coroutines aren't magic. They're implemented using state machines, and may have a number of suspension points. This is all explained in the original Coroutines KEEP.

Cancellation happens on those suspension points. A coroutine cannot be cancelled at any other point than a suspension point (in normal execution at least). If you use Thread.sleep, then there's no suspension points. You should use delay instead of sleep, which introduces a suspension point. If you're doing a long operation, you can add a few yield() to add suspension points and make your coroutine cancellable.

From the docs:

Coroutine cancellation is cooperative. A coroutine code has to cooperate to be cancellable. All the suspending functions in kotlinx.coroutines are cancellable. They check for cancellation of coroutine and throw CancellationException when cancelled. However, if a coroutine is working in a computation and does not check for cancellation, then it cannot be cancelled

Calling a suspend function automatically introduces a suspension point.

As @CommonsWare pointed out, there's is rarely a reason to create your coroutine scope. In activities and fragments, or any component tied with a lifecycle, you should use lifecycleScope. In ViewModels, there's viewModelScope.

EDIT:

I've tried adapting the source of runInterruptible to not interrupt under a certain condition: passing an instance of a custom exception class InterruptionException as a cancel cause will skip the thread interruption. I've replaced atomicfu constructs with AtomicInteger, I assumed your target was only the JVM. You'll need to opt-in internal coroutines API by adding the -Xopt-in=kotlinx.coroutines.InternalCoroutinesApi compiler flag.

suspend fun <T> runInterruptibleCancellable(
    context: CoroutineContext = EmptyCoroutineContext,
    block: () -> T
): T = withContext(context) {
    try {
        val threadState = ThreadState(coroutineContext.job)
        threadState.setup()
        try {
            block()
        } finally {
            threadState.clearInterrupt()
        }
    } catch (e: InterruptedException) {
        throw CancellationException("Blocking call was interrupted due to parent cancellation").initCause(e)
    }
}

private const val WORKING = 0
private const val FINISHED = 1
private const val INTERRUPTING = 2
private const val INTERRUPTED = 3

private class ThreadState(private val job: Job) : CompletionHandler {
    /*
       === States ===

       WORKING: running normally
       FINISH: complete normally
       INTERRUPTING: canceled, going to interrupt this thread
       INTERRUPTED: this thread is interrupted

       === Possible Transitions ===

       +----------------+         register job       +-------------------------+
       |    WORKING     |   cancellation listener    |         WORKING         |
       | (thread, null) | -------------------------> | (thread, cancel handle) |
       +----------------+                            +-------------------------+
               |                                                |   |
               | cancel                                  cancel |   | complete
               |                                                |   |
               V                                                |   |
       +---------------+                                        |   |
       | INTERRUPTING  | <--------------------------------------+   |
       +---------------+                                            |
               |                                                    |
               | interrupt                                          |
               |                                                    |
               V                                                    V
       +---------------+                              +-------------------------+
       |  INTERRUPTED  |                              |         FINISHED        |
       +---------------+                              +-------------------------+
    */
    private val _state = AtomicInteger(WORKING)
    private val targetThread = Thread.currentThread()

    // Registered cancellation handler
    private var cancelHandle: DisposableHandle? = null

    fun setup() {
        cancelHandle = job.invokeOnCompletion(onCancelling = true, invokeImmediately = true, handler = this)
        // Either we successfully stored it or it was immediately cancelled
        while (true) {
            when (val state = _state.get()) {
                // Happy-path, move forward
                WORKING -> if (_state.compareAndSet(state, WORKING)) return
                // Immediately cancelled, just continue
                INTERRUPTING, INTERRUPTED -> return
                else -> invalidState(state)
            }
        }
    }

    fun clearInterrupt() {
        /*
         * Do not allow to untriggered interrupt to leak
         */
        while (true) {
            when (val state = _state.get()) {
                WORKING -> if (_state.compareAndSet(state, FINISHED)) {
                    cancelHandle?.dispose()
                    return
                }
                INTERRUPTING -> {
                    /*
                     * Spin, cancellation mechanism is interrupting our thread right now
                     * and we have to wait it and then clear interrupt status
                     */
                }
                INTERRUPTED -> {
                    // Clear it and bail out
                    Thread.interrupted()
                    return
                }
                else -> invalidState(state)
            }
        }
    }

    // Cancellation handler
    override fun invoke(cause: Throwable?) {
        if (cause is InterruptionException) {
            while (true) {
                when (val state = _state.get()) {
                    // Working -> try to transite state and interrupt the thread
                    WORKING -> {
                        if (_state.compareAndSet(state, INTERRUPTING)) {
                            targetThread.interrupt()
                            _state.set(INTERRUPTED)
                            return
                        }
                    }
                    // Finished -- runInterruptible is already complete, INTERRUPTING - ignore
                    FINISHED, INTERRUPTING, INTERRUPTED -> return
                    else -> invalidState(state)
                }
            }
        }
    }

    private fun invalidState(state: Int): Nothing = error("Illegal state $state")
}

class InterruptionException(cause: Throwable?) : CancellationException() {
    init {
        initCause(cause)
    }
}

fun Job.interrupt(cause: Throwable? = null) {
    this.cancel(InterruptionException(cause))
}

suspend fun Job.interruptAndJoin() {
    interrupt()
    return join()
}

You can use the interrupt and interruptAndJoin extension functions to trigger thread interruption, otherwise use cancel for non-interrupting cancel. An example:

val scope = CoroutineScope(Dispatchers.IO)
val job = scope.launch {
    runInterruptibleCancellable {
        // some blocking code
        Thread.sleep(1000)
        if (!isActive) {
            println("cancelled")
        } else {
            println("completed")
        }
    }
}
job.invokeOnCompletion {
    if (it is InterruptionException) {
        print("interrupted")
    }
}

runBlocking {
//  job.interruptAndJoin()  // prints "interrupted"
//  job.cancelAndJoin()     // prints "cancelled"
    job.join()              // prints "completed"
}

This example is the only testing I've made. It seems to work. I don't know if it leaks, I don't know if it's thread-safe. I'm really far out of my expertise to be honest. Please don't use it in production without further confirmation that it works.

By default, coroutines do not do thread interrupts - as per the Making computation code cancellable documentation, using yield() or checking isActive allows coroutine aware code to participate in cancellation.

However, when interfacing with blocking code where you do want a thread interrupt, this is precisely the use case for runInterruptible(), which will cause the contained code to be thread interrupted when the coroutine scope is cancelled.

This works perfectly with lifecycle-aware coroutine scopes, which automatically cancel when the Lifecycle is destroyed:

class MainActivity : AppCompatActivity() {
    val bgDispatcher: CoroutineDispatcher = Dispatchers.IO

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)
        loadData()
    }

    private fun loadData(): Job = lifecycleScope.launch {
        Log.d("AppLog", "loadData")
        val result = runInterruptible(bgDispatcher) {
            Log.d("AppLog", "bg start")
            try {
                Thread.sleep(5000L) //this could be any executing of code, including things not editable
            } catch (e: Exception) {
                Log.d("AppLog", "$e")
            }
            Log.d("AppLog", "bg done this.isActive?${this.isActive}")
            return@runInterruptible 123
        }
        Log.d("AppLog", "got result:$result")
    }
}