Ensure that flowOn does not resume downstream after cancellation

[elizarov]

This is a problematic for Android when Main dispatcher is cancelled on destroyed activity. Atomic nature of the underlying channels is designed to prevent loss of elements, which is really not an issue for a typical application, but creates problem when used with channels.

This change introduces an optional atomic parameter to an internal ReceiveChannel.receiveOrClosed method to control cancellation atomicity of this function and tweaks emitAll(ReceiveChannel) implementation to abandon atomicity in favor of predictable cancellation.

Fixes #1265

[qwwdfsad]

Thinking about the original issue (#1265), I am concerned about the fact that other channel-related operators may expose unexpected behaviour as well (aka touching view after its destruction).
E.g. the similar test with combine operator will execute its collect block.

I'd say we should re-visit our channel-related operators, measure the performance impact of checking cancellation in collect and either do it, or do it only in channel-related operators (and, maybe, provide a checkedCollect or something like this)

[elizarov]

It's not about checking in collect. It is about checking on resume after suspension. All our suspending function do check for cancellation on resume with exception of Mutex.lock , Semaphore.acquire, and Channel.send/receive function that do not do so, we need to revisit operation that use those functions (flowOn was one of them). In fact, we should really revisit atomicity of those underlying operations in the first place.

[qwwdfsad]

Do you want to do it separately or as part of this PR?
E.g. the following code:

coroutineScope {
    val flow =  flow {
        emit(Unit) // emit to buffer
        cancel() // now cancel
    }

    flow.combine(flow) { u, _ -> u }.collect {
        // should not be reached, because cancelled before it runs
        expectUnreached()
    }
}

will throw ISE(Should not be reached)

[elizarov]

I'll fix it all as a part of this PR.

[qwwdfsad]

Looks good modulo my comment about revisiting.

[qwwdfsad]

Do you want to do it separately or as part of this PR?
E.g. the following code:

coroutineScope {
    val flow =  flow {
        emit(Unit) // emit to buffer
        cancel() // now cancel
    }

    flow.combine(flow) { u, _ -> u }.collect {
        // should not be reached, because cancelled before it runs
        expectUnreached()
    }
}

will throw ISE(Should not be reached)

[elizarov]

We need a more thorough solution as fixing all those issue one-by-one creates quite an ugly code. Here's a separate issue: #1813. Let's fix it as part of this PR, too.

[elizarov]

Aborting this PR in favor of #1937 that completely gets rid of "atomic cancellation", thus making flowOn automatically correct.