Introduce reusable cancellable continuations for hot loops with channels

[qwwdfsad]

• Average performance improvement is around 25%
• API is internal and targeted to specific usage
• DispatchedTask and DispatchedContinuation are extracted to separate files for better readability and maintainability
• Flow benchmark (temporary) rewritten to avoid take

[elizarov]

There are several problems here. First of all, we create an expensive CancelledContinuation and read state every time here, for no particular reason, since the CancelledContinuation instance will not be actually used (only cause is needed). Moreover, this cancel is also called from checkCompleted which does not need the call to cancelLater here at all.

The second problem is that direct call to continuation.cancel should not be deferred via cancelLater. The test in ReusableCancellableContinuationTest.testNotCancelledOnClaimedResume makes the wrong assumption and should be updated to reflect this fact.

Only notification from the parent via ChildContinuation class needs to go through cancelLater, so I'd suggest to remove cancelLater from here completely. Instead, create a separate function to call it from ChildContinuation:

    internal fun parentCancelled(cause: Throwable) {
        if (cancelLater(cause)) return
        cancel(cause)
    }

See a separate comment on reworking cancelLater and postponeCancellation so that it accepts cause: Throwable as parameter.

[qwwdfsad]

Nice idea with storing only the throwable.

The second problem is that direct call to continuation.cancel should not be deferred via cancelLater.

Fixed. I agree that now it is much simpler, though it now even more dangerous (moreover, it always was).

E.g. consider I am re-writing channels and now, for some reason, I start calling CancellableContinuation.cancel directly (let's omit reusability at all).

In that case, our implementation (https://github.com/Kotlin/kotlinx.coroutines/blob/master/kotlinx-coroutines-core/common/src/channels/AbstractChannel.kt#L561) will silently stop working in very rare concurrent cases.

[elizarov]

Additional note: it also makes sense to use reusable continuations in semaphore and mutex implementation. Semaphore is used in flattenMerge and it should give a bit of improvement there.

[qwwdfsad]

Done, plus found one more potential leak, added the corresponding test.
Waiting for stress tests to complete

[elizarov]

I'm happy with the general code now, sans the few comments. However, here is a general problem. This code leaks result of the last suspended coroutine for quite a prolonged time, since this result is retained in reusableCancellableContinuation.state. This might be problematic in practice. I suggest to "clean up" the result after dispatching of reusable cancellable continuation and replace it with some "RESULT" symbol to ease the rest of implementation. To further simplify code, you can always clean it up (no need to check if it is reusable or not). There's no harm to clean up result of non-reusable continuation.

The best place to do it would be in CancellableContinuationImpl.takeState, which is guaranteed to be invoked exactly once.

There's the exception of CompletedIdempotentResult. This result cannot be cleaned up. You can deal with it in a different way, though. Instead of waiting for the next attempt to reuse a continuation, you can invalidate reusability state (set DC.state to null) immediately when CompletedIdempotentResult is set in CCImpl.tryResume, thus immediately detaching CCImpl with CompletedIdempotentResult from its parent DispatchedContinuation and thus avoiding leak.

[elizarov]

Still too long.

[elizarov]

Actually, I cannot figure out an execution scenario where checkPostponedCancellation is called at state == null. It looks to to me that here we can only have REUSABLE_CLAIMED or Throwable. It cannot get invalidated while we are between claimRCC and checkPC calls and would stay either in REUSABLE_CLAIMED or Throwable state.

[qwwdfsad]

Actually, I cannot figure out an execution scenario where checkPostponedCancellation is called at state == null.

When a continuation is not reusable