diff --git a/benchmarks/src/jmh/kotlin/benchmarks/tailcall/SimpleChannel.kt b/benchmarks/src/jmh/kotlin/benchmarks/tailcall/SimpleChannel.kt index c217fcae91..d961dab8d9 100644 --- a/benchmarks/src/jmh/kotlin/benchmarks/tailcall/SimpleChannel.kt +++ b/benchmarks/src/jmh/kotlin/benchmarks/tailcall/SimpleChannel.kt @@ -70,12 +70,12 @@ class NonCancellableChannel : SimpleChannel() { } class CancellableChannel : SimpleChannel() { - override suspend fun suspendReceive(): Int = suspendAtomicCancellableCoroutine { + override suspend fun suspendReceive(): Int = suspendCancellableCoroutine { consumer = it.intercepted() COROUTINE_SUSPENDED } - override suspend fun suspendSend(element: Int) = suspendAtomicCancellableCoroutine { + override suspend fun suspendSend(element: Int) = suspendCancellableCoroutine { enqueuedValue = element producer = it.intercepted() COROUTINE_SUSPENDED @@ -84,13 +84,13 @@ class CancellableChannel : SimpleChannel() { class CancellableReusableChannel : SimpleChannel() { @Suppress("INVISIBLE_MEMBER") - override suspend fun suspendReceive(): Int = suspendAtomicCancellableCoroutineReusable { + override suspend fun suspendReceive(): Int = suspendCancellableCoroutineReusable { consumer = it.intercepted() COROUTINE_SUSPENDED } @Suppress("INVISIBLE_MEMBER") - override suspend fun suspendSend(element: Int) = suspendAtomicCancellableCoroutineReusable { + override suspend fun suspendSend(element: Int) = suspendCancellableCoroutineReusable { enqueuedValue = element producer = it.intercepted() COROUTINE_SUSPENDED diff --git a/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api b/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api index 54e355ec37..c45c60c665 100644 --- a/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api +++ b/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api @@ -80,9 +80,6 @@ public class kotlinx/coroutines/CancellableContinuationImpl : kotlin/coroutines/ public final class kotlinx/coroutines/CancellableContinuationKt { public static final fun disposeOnCancellation (Lkotlinx/coroutines/CancellableContinuation;Lkotlinx/coroutines/DisposableHandle;)V - public static final fun suspendAtomicCancellableCoroutine (Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object; - public static final fun suspendAtomicCancellableCoroutine (ZLkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object; - public static synthetic fun suspendAtomicCancellableCoroutine$default (ZLkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object; public static final fun suspendCancellableCoroutine (Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object; } diff --git a/kotlinx-coroutines-core/common/src/CancellableContinuation.kt b/kotlinx-coroutines-core/common/src/CancellableContinuation.kt index e01dc82521..3b110e9045 100644 --- a/kotlinx-coroutines-core/common/src/CancellableContinuation.kt +++ b/kotlinx-coroutines-core/common/src/CancellableContinuation.kt @@ -78,6 +78,14 @@ public interface CancellableContinuation : Continuation { @InternalCoroutinesApi public fun tryResume(value: T, idempotent: Any? = null): Any? + /** + * Same as [tryResume] but with [onCancellation] handler that called if and only if the value is not + * delivered to the caller because of the dispatch in the process, so that atomicity delivery + * guaranteed can be provided by having a cancellation fallback. + */ + @InternalCoroutinesApi + public fun tryResumeAtomic(value: T, idempotent: Any?, onCancellation: (cause: Throwable) -> Unit): Any? + /** * Tries to resume this continuation with the specified [exception] and returns a non-null object token if successful, * or `null` otherwise (it was already resumed or cancelled). When a non-null object is returned, @@ -118,8 +126,8 @@ public interface CancellableContinuation : Continuation { public fun cancel(cause: Throwable? = null): Boolean /** - * Registers a [handler] to be **synchronously** invoked on cancellation (regular or exceptional) of this continuation. - * When the continuation is already cancelled, the handler will be immediately invoked + * Registers a [handler] to be **synchronously** invoked on [cancellation][cancel] (regular or exceptional) of this continuation. + * When the continuation is already cancelled, the handler is immediately invoked * with the cancellation exception. Otherwise, the handler will be invoked as soon as this * continuation is cancelled. * @@ -128,7 +136,12 @@ public interface CancellableContinuation : Continuation { * processed as an uncaught exception in the context of the current coroutine * (see [CoroutineExceptionHandler]). * - * At most one [handler] can be installed on a continuation. + * At most one [handler] can be installed on a continuation. Attempt to call `invokeOnCancellation` second + * time produces [IllegalStateException]. + * + * This handler is also called when this continuation [resumes][resume] normally (with a value) and then + * is cancelled while waiting to be dispatched. More generally speaking, this handler is called whenever + * the caller of [suspendCancellableCoroutine] is getting a [CancellationException]. * * **Note**: Implementation of `CompletionHandler` must be fast, non-blocking, and thread-safe. * This `handler` can be invoked concurrently with the surrounding code. @@ -204,40 +217,24 @@ public suspend inline fun suspendCancellableCoroutine( } /** - * Suspends the coroutine like [suspendCancellableCoroutine], but with *atomic cancellation*. - * - * When the suspended function throws a [CancellationException], it means that the continuation was not resumed. - * As a side-effect of atomic cancellation, a thread-bound coroutine (to some UI thread, for example) may - * continue to execute even after it was cancelled from the same thread in the case when the continuation - * was already resumed and was posted for execution to the thread's queue. - * - * @suppress **This an internal API and should not be used from general code.** - */ -@InternalCoroutinesApi -public suspend inline fun suspendAtomicCancellableCoroutine( - crossinline block: (CancellableContinuation) -> Unit -): T = - suspendCoroutineUninterceptedOrReturn { uCont -> - val cancellable = CancellableContinuationImpl(uCont.intercepted(), resumeMode = MODE_ATOMIC_DEFAULT) - block(cancellable) - cancellable.getResult() - } - -/** - * Suspends coroutine similar to [suspendAtomicCancellableCoroutine], but an instance of [CancellableContinuationImpl] is reused if possible. + * Suspends the coroutine similar to [suspendCancellableCoroutine], but an instance of + * [CancellableContinuationImpl] is reused. */ -internal suspend inline fun suspendAtomicCancellableCoroutineReusable( +internal suspend inline fun suspendCancellableCoroutineReusable( crossinline block: (CancellableContinuation) -> Unit ): T = suspendCoroutineUninterceptedOrReturn { uCont -> - val cancellable = getOrCreateCancellableContinuation(uCont.intercepted()) + val cancellable = getOrCreateCancellableContinuation(uCont.intercepted(), resumeMode = MODE_CANCELLABLE_REUSABLE) block(cancellable) cancellable.getResult() } -internal fun getOrCreateCancellableContinuation(delegate: Continuation): CancellableContinuationImpl { +internal fun getOrCreateCancellableContinuation( + delegate: Continuation, resumeMode: Int +): CancellableContinuationImpl { + assert { resumeMode.isReusableMode } // If used outside of our dispatcher if (delegate !is DispatchedContinuation) { - return CancellableContinuationImpl(delegate, resumeMode = MODE_ATOMIC_DEFAULT) + return CancellableContinuationImpl(delegate, resumeMode) } /* * Attempt to claim reusable instance. @@ -253,24 +250,10 @@ internal fun getOrCreateCancellableContinuation(delegate: Continuation): * thus leaking CC instance for indefinite time. * 2) Continuation was cancelled. Then we should prevent any further reuse and bail out. */ - return delegate.claimReusableCancellableContinuation()?.takeIf { it.resetState() } - ?: return CancellableContinuationImpl(delegate, MODE_ATOMIC_DEFAULT) + return delegate.claimReusableCancellableContinuation()?.takeIf { it.resetState(resumeMode) } + ?: return CancellableContinuationImpl(delegate, resumeMode) } -/** - * @suppress **Deprecated** - */ -@Deprecated( - message = "holdCancellability parameter is deprecated and is no longer used", - replaceWith = ReplaceWith("suspendAtomicCancellableCoroutine(block)") -) -@InternalCoroutinesApi -public suspend inline fun suspendAtomicCancellableCoroutine( - holdCancellability: Boolean = false, - crossinline block: (CancellableContinuation) -> Unit -): T = - suspendAtomicCancellableCoroutine(block) - /** * Removes the specified [node] on cancellation. */ diff --git a/kotlinx-coroutines-core/common/src/CancellableContinuationImpl.kt b/kotlinx-coroutines-core/common/src/CancellableContinuationImpl.kt index 1f67dd3c6c..3ce2945524 100644 --- a/kotlinx-coroutines-core/common/src/CancellableContinuationImpl.kt +++ b/kotlinx-coroutines-core/common/src/CancellableContinuationImpl.kt @@ -88,20 +88,22 @@ internal open class CancellableContinuationImpl( private fun isReusable(): Boolean = delegate is DispatchedContinuation<*> && delegate.isReusable(this) /** - * Resets cancellability state in order to [suspendAtomicCancellableCoroutineReusable] to work. - * Invariant: used only by [suspendAtomicCancellableCoroutineReusable] in [REUSABLE_CLAIMED] state. + * Resets cancellability state in order to [suspendCancellableCoroutineReusable] to work. + * Invariant: used only by [suspendCancellableCoroutineReusable] in [REUSABLE_CLAIMED] state. */ @JvmName("resetState") // Prettier stack traces - internal fun resetState(): Boolean { + internal fun resetState(resumeMode: Int): Boolean { assert { parentHandle !== NonDisposableHandle } val state = _state.value assert { state !is NotCompleted } - if (state is CompletedIdempotentResult) { + if (state is CompletedContinuation && state.idempotentResume != null) { + // Cannot reuse continuation that was resumed with idempotent marker detachChild() return false } _decision.value = UNDECIDED _state.value = Active + this.resumeMode = resumeMode return true } @@ -129,7 +131,7 @@ internal open class CancellableContinuationImpl( private fun checkCompleted(): Boolean { val completed = isCompleted - if (resumeMode != MODE_ATOMIC_DEFAULT) return completed // Do not check postponed cancellation for non-reusable continuations + if (!resumeMode.isReusableMode) return completed // Do not check postponed cancellation for non-reusable continuations val dispatched = delegate as? DispatchedContinuation<*> ?: return completed val cause = dispatched.checkPostponedCancellation(this) ?: return completed if (!completed) { @@ -146,10 +148,21 @@ internal open class CancellableContinuationImpl( override fun takeState(): Any? = state - override fun cancelResult(state: Any?, cause: Throwable) { - if (state is CompletedWithCancellation) { - invokeHandlerSafely { - state.onCancellation(cause) + override fun cancelCompletedResult(cause: Throwable): Unit = _state.loop { state -> + when (state) { + is NotCompleted -> error("Not completed") + is CompletedExceptionally -> return // already completed exception or cancelled, nothing to do + is CompletedContinuation -> { + check(!state.cancelled) { "Must be called at most once" } + val update = state.copy(cancelCause = cause) + if (_state.compareAndSet(state, update)) { + state.invokeHandlers(this, cause) + return // done + } + } + else -> { + // completed normally without marker class, promote to CompletedContinuation to synchronize cancellation + if (_state.compareAndSet(state, CompletedContinuation(state, cancelCause = cause))) return } } } @@ -158,7 +171,7 @@ internal open class CancellableContinuationImpl( * Attempt to postpone cancellation for reusable cancellable continuation */ private fun cancelLater(cause: Throwable): Boolean { - if (resumeMode != MODE_ATOMIC_DEFAULT) return false + if (!resumeMode.isReusableMode) return false val dispatched = (delegate as? DispatchedContinuation<*>) ?: return false return dispatched.postponeCancellation(cause) } @@ -173,7 +186,7 @@ internal open class CancellableContinuationImpl( if (state is CancelHandler) invokeHandlerSafely { state.invoke(cause) } // Complete state update detachChildIfNonResuable() - dispatchResume(mode = MODE_ATOMIC_DEFAULT) + dispatchResume(mode = MODE_ATOMIC) // no need for additional cancellation checks return true } } @@ -185,7 +198,7 @@ internal open class CancellableContinuationImpl( detachChildIfNonResuable() } - private inline fun invokeHandlerSafely(block: () -> Unit) { + internal inline fun invokeHandlerSafely(block: () -> Unit) { try { block() } catch (ex: Throwable) { @@ -231,64 +244,73 @@ internal open class CancellableContinuationImpl( val state = this.state if (state is CompletedExceptionally) throw recoverStackTrace(state.cause, this) // if the parent job was already cancelled, then throw the corresponding cancellation exception - // otherwise, there is a race is suspendCancellableCoroutine { cont -> ... } does cont.resume(...) + // otherwise, there is a race if suspendCancellableCoroutine { cont -> ... } does cont.resume(...) // before the block returns. This getResult would return a result as opposed to cancellation // exception that should have happened if the continuation is dispatched for execution later. - if (resumeMode == MODE_CANCELLABLE) { + if (resumeMode.isCancellableMode) { val job = context[Job] if (job != null && !job.isActive) { val cause = job.getCancellationException() - cancelResult(state, cause) + cancelCompletedResult(cause) throw recoverStackTrace(cause, this) } } return getSuccessfulResult(state) } - override fun resumeWith(result: Result) { + override fun resumeWith(result: Result) = resumeImpl(result.toState(this), resumeMode) - } - override fun resume(value: T, onCancellation: (cause: Throwable) -> Unit) { - val cancelled = resumeImpl(CompletedWithCancellation(value, onCancellation), resumeMode) - if (cancelled != null) { - // too late to resume (was cancelled) -- call handler - invokeHandlerSafely { - onCancellation(cancelled.cause) - } - } - } + override fun resume(value: T, onCancellation: (cause: Throwable) -> Unit) = + resumeImpl(value, resumeMode, onCancellation) public override fun invokeOnCancellation(handler: CompletionHandler) { - var handleCache: CancelHandler? = null + val cancelHandler = makeCancelHandler(handler) _state.loop { state -> when (state) { is Active -> { - val node = handleCache ?: makeHandler(handler).also { handleCache = it } - if (_state.compareAndSet(state, node)) return // quit on cas success + if (_state.compareAndSet(state, cancelHandler)) return // quit on cas success } is CancelHandler -> multipleHandlersError(handler, state) - is CancelledContinuation -> { + is CompletedExceptionally -> { /* - * Continuation was already cancelled, invoke directly. + * Continuation was already cancelled or completed exceptionally. * NOTE: multiple invokeOnCancellation calls with different handlers are not allowed, - * so we check to make sure that handler was installed just once. + * so we check to make sure handler was installed just once. */ if (!state.makeHandled()) multipleHandlersError(handler, state) /* + * Call the handler only if it was cancelled (not called when completed exceptionally). * :KLUDGE: We have to invoke a handler in platform-specific way via `invokeIt` extension, * because we play type tricks on Kotlin/JS and handler is not necessarily a function there */ - invokeHandlerSafely { handler.invokeIt((state as? CompletedExceptionally)?.cause) } + if (state is CancelledContinuation) { + invokeHandlerSafely { handler.invokeIt((state as? CompletedExceptionally)?.cause) } + } return } + is CompletedContinuation -> { + /* + * Continuation was already completed, and might already have cancel handler. + */ + if (state.cancelHandler != null) multipleHandlersError(handler, state) + if (state.cancelled) { + // todo: extra layer of protection against the second invokeOnCancellation + // if (!state.makeHandled()) multipleHandlersError(handler, state) + // Was already cancelled while being dispatched -- invoke the handler directly + invokeHandlerSafely { handler.invokeIt(state.cancelCause) } + return + } + val update = state.copy(cancelHandler = cancelHandler) + if (_state.compareAndSet(state, update)) return // quit on cas success + } else -> { /* - * Continuation was already completed, do nothing. - * NOTE: multiple invokeOnCancellation calls with different handlers are not allowed, - * but we have no way to check that it was installed just once in this case. + * Continuation was already completed normally, but might get cancelled while being dispatched. + * Change its state to CompletedContinuation. */ - return + val update = CompletedContinuation(state, cancelHandler = cancelHandler) + if (_state.compareAndSet(state, update)) return // quit on cas success } } } @@ -298,7 +320,7 @@ internal open class CancellableContinuationImpl( error("It's prohibited to register multiple handlers, tried to register $handler, already has $state") } - private fun makeHandler(handler: CompletionHandler): CancelHandler = + private fun makeCancelHandler(handler: CompletionHandler): CancelHandler = if (handler is CancelHandler) handler else InvokeOnCancel(handler) private fun dispatchResume(mode: Int) { @@ -307,15 +329,38 @@ internal open class CancellableContinuationImpl( dispatch(mode) } - // returns null when successfully dispatched resumed, CancelledContinuation if too late (was already cancelled) - private fun resumeImpl(proposedUpdate: Any?, resumeMode: Int): CancelledContinuation? { + private fun resumedState( + state: NotCompleted, + proposedUpdate: Any?, + resumeMode: Int, + onCancellation: ((cause: Throwable) -> Unit)?, + idempotent: Any? + ): Any? = when { + proposedUpdate is CompletedExceptionally -> { + assert { idempotent == null } // there are no idempotent exceptional resumes + assert { onCancellation == null } // only successful results can be cancelled + proposedUpdate + } + !resumeMode.isCancellableMode && idempotent == null -> proposedUpdate // cannot be cancelled in process, all is fine + onCancellation != null || state is CancelHandler || idempotent != null -> + // mark as CompletedContinuation if special cases are present (cancellation handlers or idempotent resume) + CompletedContinuation(proposedUpdate, state as? CancelHandler, onCancellation, idempotent) + else -> proposedUpdate // simple case -- use the value directly + } + + private fun resumeImpl( + proposedUpdate: Any?, + resumeMode: Int, + onCancellation: ((cause: Throwable) -> Unit)? = null + ) { _state.loop { state -> when (state) { is NotCompleted -> { - if (!_state.compareAndSet(state, proposedUpdate)) return@loop // retry on cas failure + val update = resumedState(state, proposedUpdate, resumeMode, onCancellation, idempotent = null) + if (!_state.compareAndSet(state, update)) return@loop // retry on cas failure detachChildIfNonResuable() - dispatchResume(resumeMode) - return null + dispatchResume(resumeMode) // dispatch resume, but it might get cancelled in process + return // done } is CancelledContinuation -> { /* @@ -323,14 +368,48 @@ internal open class CancellableContinuationImpl( * because cancellation is asynchronous and may race with resume. * Racy exceptions will be lost, too. */ - if (state.makeResumed()) return state // tried to resume just once, but was cancelled + if (state.makeResumed()) { // check if trying to resume one (otherwise error) + // call onCancellation + onCancellation?.let { invokeHandlerSafely { it(state.cause) } } + return // done + } } } - alreadyResumedError(proposedUpdate) // otherwise -- an error (second resume attempt) + alreadyResumedError(proposedUpdate) // otherwise, an error (second resume attempt) } } - private fun alreadyResumedError(proposedUpdate: Any?) { + /** + * Similar to [tryResume], but does not actually completes resume (needs [completeResume] call). + * Returns [RESUME_TOKEN] when resumed, `null` when it was already resumed or cancelled. + */ + private fun tryResumeImpl( + proposedUpdate: Any?, + idempotent: Any?, + onCancellation: ((cause: Throwable) -> Unit)? + ): Symbol? { + _state.loop { state -> + when (state) { + is NotCompleted -> { + val update = resumedState(state, proposedUpdate, resumeMode, onCancellation, idempotent) + if (!_state.compareAndSet(state, update)) return@loop // retry on cas failure + detachChildIfNonResuable() + return RESUME_TOKEN + } + is CompletedContinuation -> { + return if (idempotent != null && state.idempotentResume === idempotent) { + assert { state.result === proposedUpdate } // "Non-idempotent resume" + RESUME_TOKEN // resumed with the same token -- ok + } else { + null // resumed with a different token or non-idempotent -- too late + } + } + else -> return null // cannot resume -- not active anymore + } + } + } + + private fun alreadyResumedError(proposedUpdate: Any?): Nothing { error("Already resumed, but proposed with update $proposedUpdate") } @@ -342,7 +421,7 @@ internal open class CancellableContinuationImpl( /** * Detaches from the parent. - * Invariant: used used from [CoroutineDispatcher.releaseInterceptedContinuation] iff [isReusable] is `true` + * Invariant: used from [CoroutineDispatcher.releaseInterceptedContinuation] iff [isReusable] is `true` */ internal fun detachChild() { val handle = parentHandle @@ -351,42 +430,14 @@ internal open class CancellableContinuationImpl( } // Note: Always returns RESUME_TOKEN | null - override fun tryResume(value: T, idempotent: Any?): Any? { - _state.loop { state -> - when (state) { - is NotCompleted -> { - val update: Any? = if (idempotent == null) value else - CompletedIdempotentResult(idempotent, value) - if (!_state.compareAndSet(state, update)) return@loop // retry on cas failure - detachChildIfNonResuable() - return RESUME_TOKEN - } - is CompletedIdempotentResult -> { - return if (state.idempotentResume === idempotent) { - assert { state.result === value } // "Non-idempotent resume" - RESUME_TOKEN - } else { - null - } - } - else -> return null // cannot resume -- not active anymore - } - } - } + override fun tryResume(value: T, idempotent: Any?): Any? = + tryResumeImpl(value, idempotent = idempotent, onCancellation = null) - override fun tryResumeWithException(exception: Throwable): Any? { - _state.loop { state -> - when (state) { - is NotCompleted -> { - val update = CompletedExceptionally(exception) - if (!_state.compareAndSet(state, update)) return@loop // retry on cas failure - detachChildIfNonResuable() - return RESUME_TOKEN - } - else -> return null // cannot resume -- not active anymore - } - } - } + override fun tryResumeAtomic(value: T, idempotent: Any?, onCancellation: (cause: Throwable) -> Unit): Any? = + tryResumeImpl(value, idempotent, onCancellation) + + override fun tryResumeWithException(exception: Throwable): Any? = + tryResumeImpl(CompletedExceptionally(exception), idempotent = null, onCancellation = null) // note: token is always RESUME_TOKEN override fun completeResume(token: Any) { @@ -407,8 +458,7 @@ internal open class CancellableContinuationImpl( @Suppress("UNCHECKED_CAST") override fun getSuccessfulResult(state: Any?): T = when (state) { - is CompletedIdempotentResult -> state.result as T - is CompletedWithCancellation -> state.result as T + is CompletedContinuation -> state.result as T else -> state as T } @@ -440,17 +490,18 @@ private class InvokeOnCancel( // Clashes with InvokeOnCancellation override fun toString() = "InvokeOnCancel[${handler.classSimpleName}@$hexAddress]" } -private class CompletedIdempotentResult( - @JvmField val idempotentResume: Any?, - @JvmField val result: Any? -) { - override fun toString(): String = "CompletedIdempotentResult[$result]" -} - -private class CompletedWithCancellation( +// Completed with additional metadata +private data class CompletedContinuation( @JvmField val result: Any?, - @JvmField val onCancellation: (cause: Throwable) -> Unit + @JvmField val cancelHandler: CancelHandler? = null, // installed via invokeOnCancellation + @JvmField val onCancellation: ((cause: Throwable) -> Unit)? = null, // installed via resume block + @JvmField val idempotentResume: Any? = null, + @JvmField val cancelCause: Throwable? = null ) { - override fun toString(): String = "CompletedWithCancellation[$result]" -} + val cancelled: Boolean get() = cancelCause != null + fun invokeHandlers(cont: CancellableContinuationImpl<*>, cause: Throwable) { + cancelHandler?.let { cont.invokeHandlerSafely { it.invoke(cause) } } + onCancellation?.let { cont.invokeHandlerSafely { it.invoke(cause) } } + } +} diff --git a/kotlinx-coroutines-core/common/src/channels/AbstractChannel.kt b/kotlinx-coroutines-core/common/src/channels/AbstractChannel.kt index 28c7ceabe1..5428622295 100644 --- a/kotlinx-coroutines-core/common/src/channels/AbstractChannel.kt +++ b/kotlinx-coroutines-core/common/src/channels/AbstractChannel.kt @@ -165,7 +165,7 @@ internal abstract class AbstractSendChannel : SendChannel { return closed.sendException } - private suspend fun sendSuspend(element: E): Unit = suspendAtomicCancellableCoroutineReusable sc@ { cont -> + private suspend fun sendSuspend(element: E): Unit = suspendCancellableCoroutineReusable sc@ { cont -> loop@ while (true) { if (isFullImpl) { val send = SendElement(element, cont) @@ -440,7 +440,7 @@ internal abstract class AbstractSendChannel : SendChannel { select.trySelectOther(otherOp) as Symbol? // must return symbol override fun completeResumeSend() { - block.startCoroutine(receiver = channel, completion = select.completion) + block.startCoroutineCancellable(receiver = channel, completion = select.completion) } override fun dispose() { // invoked on select completion @@ -547,7 +547,7 @@ internal abstract class AbstractChannel : AbstractSendChannel(), Channel receiveSuspend(receiveMode: Int): R = suspendAtomicCancellableCoroutineReusable sc@ { cont -> + private suspend fun receiveSuspend(receiveMode: Int): R = suspendCancellableCoroutineReusable sc@ { cont -> val receive = ReceiveElement(cont as CancellableContinuation, receiveMode) while (true) { if (enqueueReceive(receive)) { @@ -814,7 +814,7 @@ internal abstract class AbstractChannel : AbstractSendChannel(), Channel + private suspend fun hasNextSuspend(): Boolean = suspendCancellableCoroutineReusable sc@ { cont -> val receive = ReceiveHasNext(this, cont) while (true) { if (channel.enqueueReceive(receive)) { @@ -927,16 +927,16 @@ internal abstract class AbstractChannel : AbstractSendChannel(), Channel) { if (!select.trySelect()) return when (receiveMode) { RECEIVE_THROWS_ON_CLOSE -> select.resumeSelectWithException(closed.receiveException) - RECEIVE_RESULT -> block.startCoroutine(ValueOrClosed.closed(closed.closeCause), select.completion) + RECEIVE_RESULT -> block.startCoroutineCancellable(ValueOrClosed.closed(closed.closeCause), select.completion) RECEIVE_NULL_ON_CLOSE -> if (closed.closeCause == null) { - block.startCoroutine(null, select.completion) + block.startCoroutineCancellable(null, select.completion) } else { select.resumeSelectWithException(closed.receiveException) } diff --git a/kotlinx-coroutines-core/common/src/channels/Channel.kt b/kotlinx-coroutines-core/common/src/channels/Channel.kt index c4b4a9b25e..b9dd1e2577 100644 --- a/kotlinx-coroutines-core/common/src/channels/Channel.kt +++ b/kotlinx-coroutines-core/common/src/channels/Channel.kt @@ -52,12 +52,6 @@ public interface SendChannel { * This suspending function is cancellable. If the [Job] of the current coroutine is cancelled or completed while this * function is suspended, this function immediately resumes with a [CancellationException]. * - * *Cancellation of suspended `send` is atomic*: when this function - * throws a [CancellationException], it means that the [element] was not sent to this channel. - * As a side-effect of atomic cancellation, a thread-bound coroutine (to some UI thread, for example) may - * continue to execute even after it was cancelled from the same thread in the case when this `send` operation - * was already resumed and the continuation was posted for execution to the thread's queue. - * * Note that this function does not check for cancellation when it is not suspended. * Use [yield] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed. * @@ -171,12 +165,6 @@ public interface ReceiveChannel { * This suspending function is cancellable. If the [Job] of the current coroutine is cancelled or completed while this * function is suspended, this function immediately resumes with a [CancellationException]. * - * *Cancellation of suspended `receive` is atomic*: when this function - * throws a [CancellationException], it means that the element was not retrieved from this channel. - * As a side-effect of atomic cancellation, a thread-bound coroutine (to some UI thread, for example) may - * continue to execute even after it was cancelled from the same thread in the case when this `receive` operation - * was already resumed and the continuation was posted for execution to the thread's queue. - * * Note that this function does not check for cancellation when it is not suspended. * Use [yield] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed. * @@ -201,12 +189,6 @@ public interface ReceiveChannel { * This suspending function is cancellable. If the [Job] of the current coroutine is cancelled or completed while this * function is suspended, this function immediately resumes with a [CancellationException]. * - * *Cancellation of suspended `receive` is atomic*: when this function - * throws a [CancellationException], it means that the element was not retrieved from this channel. - * As a side-effect of atomic cancellation, a thread-bound coroutine (to some UI thread, for example) may - * continue to execute even after it was cancelled from the same thread in the case when this `receive` operation - * was already resumed and the continuation was posted for execution to the thread's queue. - * * Note that this function does not check for cancellation when it is not suspended. * Use [yield] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed. * @@ -251,12 +233,6 @@ public interface ReceiveChannel { * This suspending function is cancellable. If the [Job] of the current coroutine is cancelled or completed while this * function is suspended, this function immediately resumes with a [CancellationException]. * - * *Cancellation of suspended `receive` is atomic*: when this function - * throws a [CancellationException], it means that the element was not retrieved from this channel. - * As a side-effect of atomic cancellation, a thread-bound coroutine (to some UI thread, for example) may - * continue to execute even after it was cancelled from the same thread in the case when this receive operation - * was already resumed and the continuation was posted for execution to the thread's queue. - * * Note that this function does not check for cancellation when it is not suspended. * Use [yield] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed. * @@ -440,12 +416,6 @@ public interface ChannelIterator { * This suspending function is cancellable. If the [Job] of the current coroutine is cancelled or completed while this * function is suspended, this function immediately resumes with a [CancellationException]. * - * *Cancellation of suspended `receive` is atomic*: when this function - * throws a [CancellationException], it means that the element was not retrieved from this channel. - * As a side-effect of atomic cancellation, a thread-bound coroutine (to some UI thread, for example) may - * continue to execute even after it was cancelled from the same thread in the case when this receive operation - * was already resumed and the continuation was posted for execution to the thread's queue. - * * Note that this function does not check for cancellation when it is not suspended. * Use [yield] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed. */ diff --git a/kotlinx-coroutines-core/common/src/channels/Channels.common.kt b/kotlinx-coroutines-core/common/src/channels/Channels.common.kt index 8c61928aa4..13cdcd8cbb 100644 --- a/kotlinx-coroutines-core/common/src/channels/Channels.common.kt +++ b/kotlinx-coroutines-core/common/src/channels/Channels.common.kt @@ -41,12 +41,6 @@ public inline fun BroadcastChannel.consume(block: ReceiveChannel.() * This suspending function is cancellable. If the [Job] of the current coroutine is cancelled or completed while this * function is suspended, this function immediately resumes with [CancellationException]. * - * *Cancellation of suspended receive is atomic* -- when this function - * throws [CancellationException] it means that the element was not retrieved from this channel. - * As a side-effect of atomic cancellation, a thread-bound coroutine (to some UI thread, for example) may - * continue to execute even after it was cancelled from the same thread in the case when this receive operation - * was already resumed and the continuation was posted for execution to the thread's queue. - * * Note, that this function does not check for cancellation when it is not suspended. * Use [yield] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed. * diff --git a/kotlinx-coroutines-core/common/src/flow/Channels.kt b/kotlinx-coroutines-core/common/src/flow/Channels.kt index 130ffc72dc..a4e847f01c 100644 --- a/kotlinx-coroutines-core/common/src/flow/Channels.kt +++ b/kotlinx-coroutines-core/common/src/flow/Channels.kt @@ -20,6 +20,9 @@ import kotlinx.coroutines.flow.internal.unsafeFlow as flow * the channel afterwards. If you need to iterate over the channel without consuming it, * a regular `for` loop should be used instead. * + * Note, that emitting values from a channel into a flow is not atomic. A value that was received from the + * channel many not reach the flow collector if it was cancelled and will be lost. + * * This function provides a more efficient shorthand for `channel.consumeEach { value -> emit(value) }`. * See [consumeEach][ReceiveChannel.consumeEach]. */ diff --git a/kotlinx-coroutines-core/common/src/flow/operators/Context.kt b/kotlinx-coroutines-core/common/src/flow/operators/Context.kt index 4e6167f0cd..c0fe72fcb4 100644 --- a/kotlinx-coroutines-core/common/src/flow/operators/Context.kt +++ b/kotlinx-coroutines-core/common/src/flow/operators/Context.kt @@ -170,13 +170,17 @@ public fun Flow.conflate(): Flow = buffer(CONFLATED) * * For more explanation of context preservation please refer to [Flow] documentation. * - * This operators retains a _sequential_ nature of flow if changing the context does not call for changing + * This operator retains a _sequential_ nature of flow if changing the context does not call for changing * the [dispatcher][CoroutineDispatcher]. Otherwise, if changing dispatcher is required, it collects * flow emissions in one coroutine that is run using a specified [context] and emits them from another coroutines * with the original collector's context using a channel with a [default][Channel.BUFFERED] buffer size * between two coroutines similarly to [buffer] operator, unless [buffer] operator is explicitly called * before or after `flowOn`, which requests buffering behavior and specifies channel size. * + * Note, that flows operating across different dispatchers might lose some in-flight elements when cancelled. + * In particular, this operator ensures that downstream flow does not resume on cancellation even if the element + * was already emitted by the upstream flow. + * * ### Operator fusion * * Adjacent applications of [channelFlow], [flowOn], [buffer], [produceIn], and [broadcastIn] are diff --git a/kotlinx-coroutines-core/common/src/internal/DispatchedContinuation.kt b/kotlinx-coroutines-core/common/src/internal/DispatchedContinuation.kt index cf31fcf07d..45b9a232a0 100644 --- a/kotlinx-coroutines-core/common/src/internal/DispatchedContinuation.kt +++ b/kotlinx-coroutines-core/common/src/internal/DispatchedContinuation.kt @@ -19,7 +19,7 @@ internal val REUSABLE_CLAIMED = Symbol("REUSABLE_CLAIMED") internal class DispatchedContinuation( @JvmField val dispatcher: CoroutineDispatcher, @JvmField val continuation: Continuation -) : DispatchedTask(MODE_ATOMIC_DEFAULT), CoroutineStackFrame, Continuation by continuation { +) : DispatchedTask(MODE_ATOMIC), CoroutineStackFrame, Continuation by continuation { @JvmField @Suppress("PropertyName") internal var _state: Any? = UNDEFINED @@ -43,7 +43,7 @@ internal class DispatchedContinuation( * } * // state == CC * ``` - * 4) [Throwable] continuation was cancelled with this cause while being in [suspendAtomicCancellableCoroutineReusable], + * 4) [Throwable] continuation was cancelled with this cause while being in [suspendCancellableCoroutineReusable], * [CancellableContinuationImpl.getResult] will check for cancellation later. * * [REUSABLE_CLAIMED] state is required to prevent the lost resume in the channel. @@ -83,7 +83,7 @@ internal class DispatchedContinuation( } /** - * Claims the continuation for [suspendAtomicCancellableCoroutineReusable] block, + * Claims the continuation for [suspendCancellableCoroutineReusable] block, * so all cancellations will be postponed. */ @Suppress("UNCHECKED_CAST") @@ -180,10 +180,10 @@ internal class DispatchedContinuation( val state = result.toState() if (dispatcher.isDispatchNeeded(context)) { _state = state - resumeMode = MODE_ATOMIC_DEFAULT + resumeMode = MODE_ATOMIC dispatcher.dispatch(context, this) } else { - executeUnconfined(state, MODE_ATOMIC_DEFAULT) { + executeUnconfined(state, MODE_ATOMIC) { withCoroutineContext(this.context, countOrElement) { continuation.resumeWith(result) } diff --git a/kotlinx-coroutines-core/common/src/internal/DispatchedTask.kt b/kotlinx-coroutines-core/common/src/internal/DispatchedTask.kt index d11b3bd0f6..17c92397d9 100644 --- a/kotlinx-coroutines-core/common/src/internal/DispatchedTask.kt +++ b/kotlinx-coroutines-core/common/src/internal/DispatchedTask.kt @@ -8,12 +8,38 @@ import kotlinx.coroutines.internal.* import kotlin.coroutines.* import kotlin.jvm.* -@PublishedApi internal const val MODE_ATOMIC_DEFAULT = 0 // schedule non-cancellable dispatch for suspendCoroutine -@PublishedApi internal const val MODE_CANCELLABLE = 1 // schedule cancellable dispatch for suspendCancellableCoroutine -@PublishedApi internal const val MODE_UNDISPATCHED = 2 // when the thread is right, but need to mark it with current coroutine +/** + * Non-cancellable dispatch mode. + * + * **DO NOT CHANGE THE CONSTANT VALUE**. It might be inlined into legacy user code that was calling + * inline `suspendAtomicCancellableCoroutine` function and did not support reuse. + */ +internal const val MODE_ATOMIC = 0 -internal val Int.isCancellableMode get() = this == MODE_CANCELLABLE -internal val Int.isDispatchedMode get() = this == MODE_ATOMIC_DEFAULT || this == MODE_CANCELLABLE +/** + * Cancellable dispatch mode. It is used by user-facing [suspendCancellableCoroutine]. + * Note, that implementation of cancellability checks mode via [Int.isCancellableMode] extension. + * + * **DO NOT CHANGE THE CONSTANT VALUE**. It is being into the user code from [suspendCancellableCoroutine]. + */ +@PublishedApi +internal const val MODE_CANCELLABLE = 1 + +/** + * Cancellable dispatch mode for [suspendCancellableCoroutineReusable]. + * Note, that implementation of cancellability checks mode via [Int.isCancellableMode] extension; + * implementation of reuse checks mode via [Int.isReusableMode] extension. + */ +internal const val MODE_CANCELLABLE_REUSABLE = 2 + +/** + * Undispatched mode for [CancellableContinuation.resumeUndispatched]. + * It is used when the thread is right, but it needs to be mark it with the current coroutine. + */ +internal const val MODE_UNDISPATCHED = 4 + +internal val Int.isCancellableMode get() = this == MODE_CANCELLABLE || this == MODE_CANCELLABLE_REUSABLE +internal val Int.isReusableMode get() = this == MODE_CANCELLABLE_REUSABLE internal abstract class DispatchedTask( @JvmField public var resumeMode: Int @@ -22,7 +48,10 @@ internal abstract class DispatchedTask( internal abstract fun takeState(): Any? - internal open fun cancelResult(state: Any?, cause: Throwable) {} + /** + * Called when this task was cancelled while it was being dispatched. + */ + internal open fun cancelCompletedResult(cause: Throwable) {} @Suppress("UNCHECKED_CAST") internal open fun getSuccessfulResult(state: Any?): T = @@ -49,7 +78,7 @@ internal abstract class DispatchedTask( val job = if (exception == null && resumeMode.isCancellableMode) context[Job] else null if (job != null && !job.isActive) { val cause = job.getCancellationException() - cancelResult(state, cause) + cancelCompletedResult(cause) continuation.resumeWithStackTrace(cause) } else { if (exception != null) { @@ -101,7 +130,8 @@ internal abstract class DispatchedTask( internal fun DispatchedTask.dispatch(mode: Int) { val delegate = this.delegate - if (mode.isDispatchedMode && delegate is DispatchedContinuation<*> && mode.isCancellableMode == resumeMode.isCancellableMode) { + val undispatched = mode == MODE_UNDISPATCHED + if (!undispatched && delegate is DispatchedContinuation<*> && mode.isCancellableMode == resumeMode.isCancellableMode) { // dispatch directly using this instance's Runnable implementation val dispatcher = delegate.dispatcher val context = delegate.context @@ -111,21 +141,21 @@ internal fun DispatchedTask.dispatch(mode: Int) { resumeUnconfined() } } else { - resume(delegate, mode) + // delegate is coming from 3rd-party interceptor implementation (and does not support cancellation) + // or undispatched mode was requested + resume(delegate, undispatched) } } @Suppress("UNCHECKED_CAST") -internal fun DispatchedTask.resume(delegate: Continuation, useMode: Int) { - // slow-path - use delegate +internal fun DispatchedTask.resume(delegate: Continuation, undispatched: Boolean) { + // This resume is never cancellable. The result is always delivered to delegate continuation. val state = takeState() val exception = getExceptionalResult(state)?.let { recoverStackTrace(it, delegate) } - val result = if (exception != null) Result.failure(exception) else Result.success(state as T) - when (useMode) { - MODE_ATOMIC_DEFAULT -> delegate.resumeWith(result) - MODE_CANCELLABLE -> delegate.resumeCancellableWith(result) - MODE_UNDISPATCHED -> (delegate as DispatchedContinuation).resumeUndispatchedWith(result) - else -> error("Invalid mode $useMode") + val result = if (exception != null) Result.failure(exception) else Result.success(getSuccessfulResult(state)) + when { + undispatched -> (delegate as DispatchedContinuation).resumeUndispatchedWith(result) + else -> delegate.resumeWith(result) } } @@ -137,7 +167,7 @@ private fun DispatchedTask<*>.resumeUnconfined() { } else { // Was not active -- run event loop until all unconfined tasks are executed runUnconfinedEventLoop(eventLoop) { - resume(delegate, MODE_UNDISPATCHED) + resume(delegate, undispatched = true) } } } diff --git a/kotlinx-coroutines-core/common/src/selects/Select.kt b/kotlinx-coroutines-core/common/src/selects/Select.kt index 0595341f92..d4b5ca1320 100644 --- a/kotlinx-coroutines-core/common/src/selects/Select.kt +++ b/kotlinx-coroutines-core/common/src/selects/Select.kt @@ -188,14 +188,6 @@ public interface SelectInstance { * This suspending function is cancellable. If the [Job] of the current coroutine is cancelled or completed while this * function is suspended, this function immediately resumes with [CancellationException]. * - * Atomicity of cancellation depends on the clause: [onSend][SendChannel.onSend], [onReceive][ReceiveChannel.onReceive], - * [onReceiveOrNull][ReceiveChannel.onReceiveOrNull], and [onLock][Mutex.onLock] clauses are - * *atomically cancellable*. When select throws [CancellationException] it means that those clauses had not performed - * their respective operations. - * As a side-effect of atomic cancellation, a thread-bound coroutine (to some UI thread, for example) may - * continue to execute even after it was cancelled from the same thread in the case when this select operation - * was already resumed on atomically cancellable clause and the continuation was posted for execution to the thread's queue. - * * Note that this function does not check for cancellation when it is not suspended. * Use [yield] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed. */ diff --git a/kotlinx-coroutines-core/common/src/sync/Mutex.kt b/kotlinx-coroutines-core/common/src/sync/Mutex.kt index 1b11bc96cc..c482b7004f 100644 --- a/kotlinx-coroutines-core/common/src/sync/Mutex.kt +++ b/kotlinx-coroutines-core/common/src/sync/Mutex.kt @@ -44,12 +44,7 @@ public interface Mutex { * * This suspending function is cancellable. If the [Job] of the current coroutine is cancelled or completed while this * function is suspended, this function immediately resumes with [CancellationException]. - * - * *Cancellation of suspended lock invocation is atomic* -- when this function - * throws [CancellationException] it means that the mutex was not locked. - * As a side-effect of atomic cancellation, a thread-bound coroutine (to some UI thread, for example) may - * continue to execute even after it was cancelled from the same thread in the case when this lock operation - * was already resumed and the continuation was posted for execution to the thread's queue. + * The lock is not acquired if [CancellationException] was thrown. * * Note that this function does not check for cancellation when it is not suspended. * Use [yield] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed. @@ -118,8 +113,6 @@ public suspend inline fun Mutex.withLock(owner: Any? = null, action: () -> T @SharedImmutable private val LOCK_FAIL = Symbol("LOCK_FAIL") @SharedImmutable -private val ENQUEUE_FAIL = Symbol("ENQUEUE_FAIL") -@SharedImmutable private val UNLOCK_FAIL = Symbol("UNLOCK_FAIL") @SharedImmutable private val SELECT_SUCCESS = Symbol("SELECT_SUCCESS") @@ -188,8 +181,8 @@ internal class MutexImpl(locked: Boolean) : Mutex, SelectClause2 { return lockSuspend(owner) } - private suspend fun lockSuspend(owner: Any?) = suspendAtomicCancellableCoroutineReusable sc@ { cont -> - val waiter = LockCont(owner, cont) + private suspend fun lockSuspend(owner: Any?) = suspendCancellableCoroutineReusable sc@ { cont -> + val waiter = LockCont(this, owner, cont) _state.loop { state -> when (state) { is Empty -> { @@ -248,7 +241,7 @@ internal class MutexImpl(locked: Boolean) : Mutex, SelectClause2 { } is LockedQueue -> { check(state.owner !== owner) { "Already locked by $owner" } - val node = LockSelect(owner, this, select, block) + val node = LockSelect(this, owner, select, block) if (state.addLastIf(node) { _state.value === state }) { // successfully enqueued select.disposeOnSelect(node) @@ -347,6 +340,7 @@ internal class MutexImpl(locked: Boolean) : Mutex, SelectClause2 { } private abstract class LockWaiter( + @JvmField val mutex: Mutex, @JvmField val owner: Any? ) : LockFreeLinkedListNode(), DisposableHandle { final override fun dispose() { remove() } @@ -355,24 +349,28 @@ internal class MutexImpl(locked: Boolean) : Mutex, SelectClause2 { } private class LockCont( + mutex: MutexImpl, owner: Any?, @JvmField val cont: CancellableContinuation - ) : LockWaiter(owner) { - override fun tryResumeLockWaiter() = cont.tryResume(Unit) + ) : LockWaiter(mutex, owner) { + override fun tryResumeLockWaiter() = cont.tryResumeAtomic(Unit, idempotent = null) { + // if this continuation get's cancelled during dispatch to the caller, then release the lock + mutex.unlock(owner) + } override fun completeResumeLockWaiter(token: Any) = cont.completeResume(token) override fun toString(): String = "LockCont[$owner, $cont]" } private class LockSelect( + mutex: Mutex, owner: Any?, - @JvmField val mutex: Mutex, @JvmField val select: SelectInstance, @JvmField val block: suspend (Mutex) -> R - ) : LockWaiter(owner) { + ) : LockWaiter(mutex, owner) { override fun tryResumeLockWaiter(): Any? = if (select.trySelect()) SELECT_SUCCESS else null override fun completeResumeLockWaiter(token: Any) { assert { token === SELECT_SUCCESS } - block.startCoroutine(receiver = mutex, completion = select.completion) + block.startCoroutineCancellable(receiver = mutex, completion = select.completion) } override fun toString(): String = "LockSelect[$owner, $mutex, $select]" } diff --git a/kotlinx-coroutines-core/common/src/sync/Semaphore.kt b/kotlinx-coroutines-core/common/src/sync/Semaphore.kt index aa7ed63d3d..f342c4e619 100644 --- a/kotlinx-coroutines-core/common/src/sync/Semaphore.kt +++ b/kotlinx-coroutines-core/common/src/sync/Semaphore.kt @@ -8,7 +8,6 @@ import kotlinx.atomicfu.* import kotlinx.coroutines.* import kotlinx.coroutines.internal.* import kotlin.coroutines.* -import kotlin.jvm.* import kotlin.math.* import kotlin.native.concurrent.* @@ -33,9 +32,7 @@ public interface Semaphore { * * This suspending function is cancellable. If the [Job] of the current coroutine is cancelled or completed while this * function is suspended, this function immediately resumes with [CancellationException]. - * - * *Cancellation of suspended semaphore acquisition is atomic* -- when this function - * throws [CancellationException] it means that the semaphore was not acquired. + * The semaphore is not acquired if [CancellationException] was thrown. * * Note, that this function does not check for cancellation when it does not suspend. * Use [CoroutineScope.isActive] or [CoroutineScope.ensureActive] to periodically @@ -136,7 +133,7 @@ private class SemaphoreImpl( cur + 1 } - private suspend fun addToQueueAndSuspend() = suspendAtomicCancellableCoroutineReusable sc@ { cont -> + private suspend fun addToQueueAndSuspend() = suspendCancellableCoroutineReusable sc@ { cont -> val last = this.tail val enqIdx = enqIdx.getAndIncrement() val segment = getSegment(last, enqIdx / SEGMENT_SIZE) diff --git a/kotlinx-coroutines-core/common/test/AtomicCancellationCommonTest.kt b/kotlinx-coroutines-core/common/test/AtomicCancellationCommonTest.kt index a9f58dd6ee..c763faf225 100644 --- a/kotlinx-coroutines-core/common/test/AtomicCancellationCommonTest.kt +++ b/kotlinx-coroutines-core/common/test/AtomicCancellationCommonTest.kt @@ -87,23 +87,23 @@ class AtomicCancellationCommonTest : TestBase() { } @Test - fun testLockAtomicCancel() = runTest { + fun testLockCancellable() = runTest { expect(1) val mutex = Mutex(true) // locked mutex val job = launch(start = CoroutineStart.UNDISPATCHED) { expect(2) mutex.lock() // suspends - expect(4) // should execute despite cancellation + expectUnreached() // should NOT execute because of cancellation } expect(3) mutex.unlock() // unlock mutex first job.cancel() // cancel the job next yield() // now yield - finish(5) + finish(4) } @Test - fun testSelectLockAtomicCancel() = runTest { + fun testSelectLockCancellable() = runTest { expect(1) val mutex = Mutex(true) // locked mutex val job = launch(start = CoroutineStart.UNDISPATCHED) { @@ -114,13 +114,12 @@ class AtomicCancellationCommonTest : TestBase() { "OK" } } - assertEquals("OK", result) - expect(5) // should execute despite cancellation + expectUnreached() // should NOT execute because of cancellation } expect(3) mutex.unlock() // unlock mutex first job.cancel() // cancel the job next yield() // now yield - finish(6) + finish(4) } } \ No newline at end of file diff --git a/kotlinx-coroutines-core/common/test/CancellableContinuationHandlersTest.kt b/kotlinx-coroutines-core/common/test/CancellableContinuationHandlersTest.kt index 00f719e632..bbdd0d0412 100644 --- a/kotlinx-coroutines-core/common/test/CancellableContinuationHandlersTest.kt +++ b/kotlinx-coroutines-core/common/test/CancellableContinuationHandlersTest.kt @@ -23,10 +23,23 @@ class CancellableContinuationHandlersTest : TestBase() { fun testDoubleSubscriptionAfterCompletion() = runTest { suspendCancellableCoroutine { c -> c.resume(Unit) - // Nothing happened - c.invokeOnCancellation { expectUnreached() } - // Cannot validate after completion + // First invokeOnCancellation is Ok c.invokeOnCancellation { expectUnreached() } + // Second invokeOnCancellation is not allowed + assertFailsWith { c.invokeOnCancellation { expectUnreached() } } + } + } + + @Test + fun testDoubleSubscriptionAfterCompletionWithException() = runTest { + assertFailsWith { + suspendCancellableCoroutine { c -> + c.resumeWithException(TestException()) + // First invokeOnCancellation is Ok + c.invokeOnCancellation { expectUnreached() } + // Second invokeOnCancellation is not allowed + assertFailsWith { c.invokeOnCancellation { expectUnreached() } } + } } } diff --git a/kotlinx-coroutines-core/common/test/CancellableResumeTest.kt b/kotlinx-coroutines-core/common/test/CancellableResumeTest.kt index b2cde6b978..d89325d07c 100644 --- a/kotlinx-coroutines-core/common/test/CancellableResumeTest.kt +++ b/kotlinx-coroutines-core/common/test/CancellableResumeTest.kt @@ -99,24 +99,29 @@ class CancellableResumeTest : TestBase() { try { suspendCancellableCoroutine { cont -> expect(3) - // resumed first, then cancelled, so no invokeOnCancellation call - cont.invokeOnCancellation { expectUnreached() } + // resumed first, dispatched, then cancelled, bit still got invokeOnCancellation call + cont.invokeOnCancellation { cause -> + // Note: invokeOnCancellation is called before cc.resume(value) { ... } handler + expect(7) + assertTrue(cause is TestCancellationException) + } cc = cont } expectUnreached() } catch (e: CancellationException) { - expect(8) + expect(9) } } expect(4) cc.resume("OK") { cause -> - expect(7) + // Note: this handler is called after invokeOnCancellation handler + expect(8) assertTrue(cause is TestCancellationException) } expect(5) job.cancel(TestCancellationException()) // cancel while execution is dispatched expect(6) yield() // to coroutine -- throws cancellation exception - finish(9) + finish(10) } } \ No newline at end of file diff --git a/kotlinx-coroutines-core/common/test/channels/BroadcastTest.kt b/kotlinx-coroutines-core/common/test/channels/BroadcastTest.kt index bb3142e54c..ab1a85d697 100644 --- a/kotlinx-coroutines-core/common/test/channels/BroadcastTest.kt +++ b/kotlinx-coroutines-core/common/test/channels/BroadcastTest.kt @@ -63,7 +63,7 @@ class BroadcastTest : TestBase() { val a = produce { expect(3) send("MSG") - expect(5) + expectUnreached() // is not executed, because send is cancelled } expect(2) yield() // to produce @@ -72,7 +72,7 @@ class BroadcastTest : TestBase() { expect(4) yield() // to abort produce assertTrue(a.isClosedForReceive) // the source channel was consumed - finish(6) + finish(5) } @Test diff --git a/kotlinx-coroutines-core/common/test/flow/operators/CatchTest.kt b/kotlinx-coroutines-core/common/test/flow/operators/CatchTest.kt index 802ba1ef2f..eedfac2ea3 100644 --- a/kotlinx-coroutines-core/common/test/flow/operators/CatchTest.kt +++ b/kotlinx-coroutines-core/common/test/flow/operators/CatchTest.kt @@ -134,15 +134,14 @@ class CatchTest : TestBase() { // flowOn with a different dispatcher introduces asynchrony so that all exceptions in the // upstream flows are handled before they go downstream .onEach { value -> - expect(8) - assertEquals("OK", value) + expectUnreached() // already cancelled } .catch { e -> - expect(9) + expect(8) assertTrue(e is TestException) assertSame(d0, kotlin.coroutines.coroutineContext[ContinuationInterceptor] as CoroutineContext) } .collect() - finish(10) + finish(9) } } diff --git a/kotlinx-coroutines-core/common/test/flow/operators/CombineTest.kt b/kotlinx-coroutines-core/common/test/flow/operators/CombineTest.kt index a619355b68..2893321998 100644 --- a/kotlinx-coroutines-core/common/test/flow/operators/CombineTest.kt +++ b/kotlinx-coroutines-core/common/test/flow/operators/CombineTest.kt @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license. + * Copyright 2016-2020 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license. */ package kotlinx.coroutines.flow @@ -238,6 +238,22 @@ abstract class CombineTestBase : TestBase() { assertFailsWith(flow) finish(7) } + + @Test + fun testCancelledCombine() = runTest( + expected = { it is CancellationException } + ) { + coroutineScope { + val flow = flow { + emit(Unit) // emit + } + cancel() // cancel the scope + flow.combineLatest(flow) { u, _ -> u }.collect { + // should not be reached, because cancelled before it runs + expectUnreached() + } + } + } } class CombineTest : CombineTestBase() { diff --git a/kotlinx-coroutines-core/common/test/flow/operators/FlowOnTest.kt b/kotlinx-coroutines-core/common/test/flow/operators/FlowOnTest.kt index f8350ff584..0eae1a3860 100644 --- a/kotlinx-coroutines-core/common/test/flow/operators/FlowOnTest.kt +++ b/kotlinx-coroutines-core/common/test/flow/operators/FlowOnTest.kt @@ -341,4 +341,19 @@ class FlowOnTest : TestBase() { assertEquals(expected, value) } } + + @Test + fun testCancelledFlowOn() = runTest { + assertFailsWith { + coroutineScope { + flow { + emit(Unit) // emit to buffer + cancel() // now cancel + }.flowOn(wrapperDispatcher()).collect { + // should not be reached, because cancelled before it runs + expectUnreached() + } + } + } + } } diff --git a/kotlinx-coroutines-core/common/test/flow/operators/ZipTest.kt b/kotlinx-coroutines-core/common/test/flow/operators/ZipTest.kt index b28320c391..5f2b5a74cd 100644 --- a/kotlinx-coroutines-core/common/test/flow/operators/ZipTest.kt +++ b/kotlinx-coroutines-core/common/test/flow/operators/ZipTest.kt @@ -67,14 +67,12 @@ class ZipTest : TestBase() { val f1 = flow { emit("1") emit("2") - hang { - expect(1) - } + expectUnreached() // the above emit will get cancelled because f2 ends } val f2 = flowOf("a", "b") assertEquals(listOf("1a", "2b"), f1.zip(f2) { s1, s2 -> s1 + s2 }.toList()) - finish(2) + finish(1) } @Test @@ -92,25 +90,6 @@ class ZipTest : TestBase() { finish(2) } - @Test - fun testCancelWhenFlowIsDone2() = runTest { - val f1 = flow { - emit("1") - emit("2") - try { - emit("3") - expectUnreached() - } finally { - expect(1) - } - - } - - val f2 = flowOf("a", "b") - assertEquals(listOf("1a", "2b"), f1.zip(f2) { s1, s2 -> s1 + s2 }.toList()) - finish(2) - } - @Test fun testContextIsIsolatedReversed() = runTest { val f1 = flow { diff --git a/kotlinx-coroutines-core/common/test/selects/SelectLoopTest.kt b/kotlinx-coroutines-core/common/test/selects/SelectLoopTest.kt index 5af68f6be5..e31ccfc16d 100644 --- a/kotlinx-coroutines-core/common/test/selects/SelectLoopTest.kt +++ b/kotlinx-coroutines-core/common/test/selects/SelectLoopTest.kt @@ -24,19 +24,20 @@ class SelectLoopTest : TestBase() { expect(3) throw TestException() } - var isDone = false - while (!isDone) { - select { - channel.onReceiveOrNull { - expect(4) - assertEquals(Unit, it) - } - job.onJoin { - expect(5) - isDone = true + try { + while (true) { + select { + channel.onReceiveOrNull { + expectUnreached() + } + job.onJoin { + expectUnreached() + } } } + } catch (e: CancellationException) { + // select will get cancelled because of the failure of job + finish(4) } - finish(6) } } \ No newline at end of file diff --git a/kotlinx-coroutines-core/jvm/test/AtomicCancellationTest.kt b/kotlinx-coroutines-core/jvm/test/AtomicCancellationTest.kt index 8a7dce01ee..2612b84153 100644 --- a/kotlinx-coroutines-core/jvm/test/AtomicCancellationTest.kt +++ b/kotlinx-coroutines-core/jvm/test/AtomicCancellationTest.kt @@ -9,25 +9,24 @@ import kotlinx.coroutines.selects.* import kotlin.test.* class AtomicCancellationTest : TestBase() { - @Test - fun testSendAtomicCancel() = runBlocking { + fun testSendCancellable() = runBlocking { expect(1) val channel = Channel() val job = launch(start = CoroutineStart.UNDISPATCHED) { expect(2) channel.send(42) // suspends - expect(4) // should execute despite cancellation + expectUnreached() // should NOT execute because of cancellation } expect(3) assertEquals(42, channel.receive()) // will schedule sender for further execution job.cancel() // cancel the job next yield() // now yield - finish(5) + finish(4) } @Test - fun testSelectSendAtomicCancel() = runBlocking { + fun testSelectSendCancellable() = runBlocking { expect(1) val channel = Channel() val job = launch(start = CoroutineStart.UNDISPATCHED) { @@ -38,34 +37,33 @@ class AtomicCancellationTest : TestBase() { "OK" } } - assertEquals("OK", result) - expect(5) // should execute despite cancellation + expectUnreached() // should NOT execute because of cancellation } expect(3) assertEquals(42, channel.receive()) // will schedule sender for further execution job.cancel() // cancel the job next yield() // now yield - finish(6) + finish(4) } @Test - fun testReceiveAtomicCancel() = runBlocking { + fun testReceiveCancellable() = runBlocking { expect(1) val channel = Channel() val job = launch(start = CoroutineStart.UNDISPATCHED) { expect(2) assertEquals(42, channel.receive()) // suspends - expect(4) // should execute despite cancellation + expectUnreached() // should NOT execute because of cancellation } expect(3) channel.send(42) // will schedule receiver for further execution job.cancel() // cancel the job next yield() // now yield - finish(5) + finish(4) } @Test - fun testSelectReceiveAtomicCancel() = runBlocking { + fun testSelectReceiveCancellable() = runBlocking { expect(1) val channel = Channel() val job = launch(start = CoroutineStart.UNDISPATCHED) { @@ -77,14 +75,13 @@ class AtomicCancellationTest : TestBase() { "OK" } } - assertEquals("OK", result) - expect(5) // should execute despite cancellation + expectUnreached() // should NOT execute because of cancellation } expect(3) channel.send(42) // will schedule receiver for further execution job.cancel() // cancel the job next yield() // now yield - finish(6) + finish(4) } @Test diff --git a/kotlinx-coroutines-core/jvm/test/JobStructuredJoinStressTest.kt b/kotlinx-coroutines-core/jvm/test/JobStructuredJoinStressTest.kt index ec3635ca36..50d86f32be 100644 --- a/kotlinx-coroutines-core/jvm/test/JobStructuredJoinStressTest.kt +++ b/kotlinx-coroutines-core/jvm/test/JobStructuredJoinStressTest.kt @@ -5,6 +5,7 @@ package kotlinx.coroutines import org.junit.* +import kotlin.coroutines.* /** * Test a race between job failure and join. @@ -12,22 +13,52 @@ import org.junit.* * See [#1123](https://github.com/Kotlin/kotlinx.coroutines/issues/1123). */ class JobStructuredJoinStressTest : TestBase() { - private val nRepeats = 1_000 * stressTestMultiplier + private val nRepeats = 10_000 * stressTestMultiplier @Test - fun testStress() { - repeat(nRepeats) { + fun testStressRegularJoin() { + stress(Job::join) + } + + @Test + fun testStressSuspendCancellable() { + stress { job -> + suspendCancellableCoroutine { cont -> + job.invokeOnCompletion { cont.resume(Unit) } + } + } + } + + @Test + fun testStressSuspendCancellableReusable() { + stress { job -> + suspendCancellableCoroutineReusable { cont -> + job.invokeOnCompletion { cont.resume(Unit) } + } + } + } + + private fun stress(join: suspend (Job) -> Unit) { + expect(1) + repeat(nRepeats) { index -> assertFailsWith { runBlocking { // launch in background val job = launch(Dispatchers.Default) { throw TestException("OK") // crash } - assertFailsWith { - job.join() + try { + join(job) + error("Should not complete successfully") + } catch (e: CancellationException) { + // must always crash with cancellation exception + expect(2 + index) + } catch (e: Throwable) { + error("Unexpected exception", e) } } } } + finish(2 + nRepeats) } } \ No newline at end of file diff --git a/kotlinx-coroutines-core/jvm/test/ReusableCancellableContinuationTest.kt b/kotlinx-coroutines-core/jvm/test/ReusableCancellableContinuationTest.kt index 5f5620c632..c8a94b682a 100644 --- a/kotlinx-coroutines-core/jvm/test/ReusableCancellableContinuationTest.kt +++ b/kotlinx-coroutines-core/jvm/test/ReusableCancellableContinuationTest.kt @@ -11,15 +11,14 @@ import kotlin.coroutines.* import kotlin.test.* class ReusableCancellableContinuationTest : TestBase() { - @Test fun testReusable() = runTest { - testContinuationsCount(10, 1, ::suspendAtomicCancellableCoroutineReusable) + testContinuationsCount(10, 1, ::suspendCancellableCoroutineReusable) } @Test fun testRegular() = runTest { - testContinuationsCount(10, 10, ::suspendAtomicCancellableCoroutine) + testContinuationsCount(10, 10, ::suspendCancellableCoroutine) } private suspend inline fun CoroutineScope.testContinuationsCount( @@ -51,7 +50,7 @@ class ReusableCancellableContinuationTest : TestBase() { fun testCancelledOnClaimedCancel() = runTest { expect(1) try { - suspendAtomicCancellableCoroutineReusable { + suspendCancellableCoroutineReusable { it.cancel() } expectUnreached() @@ -65,7 +64,7 @@ class ReusableCancellableContinuationTest : TestBase() { expect(1) // Bind child at first var continuation: Continuation<*>? = null - suspendAtomicCancellableCoroutineReusable { + suspendCancellableCoroutineReusable { expect(2) continuation = it launch { // Attach to the parent, avoid fast path @@ -77,13 +76,16 @@ class ReusableCancellableContinuationTest : TestBase() { ensureActive() // Verify child was bound FieldWalker.assertReachableCount(1, coroutineContext[Job]) { it === continuation } - suspendAtomicCancellableCoroutineReusable { - expect(5) - coroutineContext[Job]!!.cancel() - it.resume(Unit) + try { + suspendCancellableCoroutineReusable { + expect(5) + coroutineContext[Job]!!.cancel() + it.resume(Unit) // will not dispatch, will get CancellationException + } + } catch (e: CancellationException) { + assertFalse(isActive) + finish(6) } - assertFalse(isActive) - finish(6) } @Test @@ -93,7 +95,7 @@ class ReusableCancellableContinuationTest : TestBase() { launch { cont!!.resumeWith(Result.success(Unit)) } - suspendAtomicCancellableCoroutineReusable { + suspendCancellableCoroutineReusable { cont = it } ensureActive() @@ -108,7 +110,7 @@ class ReusableCancellableContinuationTest : TestBase() { launch { // Attach to the parent, avoid fast path cont!!.resumeWith(Result.success(Unit)) } - suspendAtomicCancellableCoroutine { + suspendCancellableCoroutine { cont = it } ensureActive() @@ -121,7 +123,7 @@ class ReusableCancellableContinuationTest : TestBase() { expect(1) var cont: Continuation<*>? = null try { - suspendAtomicCancellableCoroutineReusable { + suspendCancellableCoroutineReusable { cont = it it.cancel() } @@ -137,7 +139,7 @@ class ReusableCancellableContinuationTest : TestBase() { val currentJob = coroutineContext[Job]!! expect(1) // Bind child at first - suspendAtomicCancellableCoroutineReusable { + suspendCancellableCoroutineReusable { expect(2) // Attach to the parent, avoid fast path launch { @@ -153,15 +155,23 @@ class ReusableCancellableContinuationTest : TestBase() { assertFalse(isActive) // Child detached FieldWalker.assertReachableCount(0, currentJob) { it is CancellableContinuation<*> } - suspendAtomicCancellableCoroutineReusable { it.resume(Unit) } - suspendAtomicCancellableCoroutineReusable { it.resume(Unit) } - FieldWalker.assertReachableCount(0, currentJob) { it is CancellableContinuation<*> } - + expect(5) + try { + // Resume is non-atomic, so it throws cancellation exception + suspendCancellableCoroutineReusable { + expect(6) // but the code inside the block is executed + it.resume(Unit) + } + } catch (e: CancellationException) { + FieldWalker.assertReachableCount(0, currentJob) { it is CancellableContinuation<*> } + expect(7) + } try { - suspendAtomicCancellableCoroutineReusable {} + // No resume -- still cancellation exception + suspendCancellableCoroutineReusable {} } catch (e: CancellationException) { FieldWalker.assertReachableCount(0, currentJob) { it is CancellableContinuation<*> } - finish(5) + finish(8) } } diff --git a/kotlinx-coroutines-core/jvm/test/channels/BroadcastChannelMultiReceiveStressTest.kt b/kotlinx-coroutines-core/jvm/test/channels/BroadcastChannelMultiReceiveStressTest.kt index 54ba7b639f..2e73b2432a 100644 --- a/kotlinx-coroutines-core/jvm/test/channels/BroadcastChannelMultiReceiveStressTest.kt +++ b/kotlinx-coroutines-core/jvm/test/channels/BroadcastChannelMultiReceiveStressTest.kt @@ -48,8 +48,9 @@ class BroadcastChannelMultiReceiveStressTest( launch(pool + CoroutineName("Sender")) { var i = 0L while (isActive) { - broadcast.send(++i) - sentTotal.set(i) // set sentTotal only if `send` was not cancelled + i++ + broadcast.send(i) // could be cancelled + sentTotal.set(i) // only was for it if it was not cancelled } } val receivers = mutableListOf() @@ -88,10 +89,8 @@ class BroadcastChannelMultiReceiveStressTest( try { withTimeout(5000) { receivers.forEachIndexed { index, receiver -> - if (lastReceived[index].get() == total) - receiver.cancel() - else - receiver.join() + if (lastReceived[index].get() >= total) receiver.cancel() + receiver.join() } } } catch (e: Exception) { @@ -112,7 +111,7 @@ class BroadcastChannelMultiReceiveStressTest( check(i == last + 1) { "Last was $last, got $i" } receivedTotal.incrementAndGet() lastReceived[receiverIndex].set(i) - return i == stopOnReceive.get() + return i >= stopOnReceive.get() } private suspend fun doReceive(channel: ReceiveChannel, receiverIndex: Int) { diff --git a/kotlinx-coroutines-core/jvm/test/sync/MutexStressTest.kt b/kotlinx-coroutines-core/jvm/test/sync/MutexStressTest.kt index 8ecb8fd741..2fe9f7a178 100644 --- a/kotlinx-coroutines-core/jvm/test/sync/MutexStressTest.kt +++ b/kotlinx-coroutines-core/jvm/test/sync/MutexStressTest.kt @@ -26,4 +26,34 @@ class MutexStressTest : TestBase() { jobs.forEach { it.join() } assertEquals(n * k, shared) } + + @Test + fun stressUnlockCancelRace() = runTest { + val n = 10_000 * stressTestMultiplier + val mutex = Mutex(true) // create a locked mutex + newSingleThreadContext("SemaphoreStressTest").use { pool -> + repeat (n) { + // Initially, we hold the lock and no one else can `lock`, + // otherwise it's a bug. + assertTrue(mutex.isLocked) + var job1EnteredCriticalSection = false + val job1 = launch(start = CoroutineStart.UNDISPATCHED) { + mutex.lock() + job1EnteredCriticalSection = true + mutex.unlock() + } + // check that `job1` didn't finish the call to `acquire()` + assertEquals(false, job1EnteredCriticalSection) + val job2 = launch(pool) { + mutex.unlock() + } + // Because `job2` executes in a separate thread, this + // cancellation races with the call to `unlock()`. + job1.cancelAndJoin() + job2.join() + assertFalse(mutex.isLocked) + mutex.lock() + } + } + } } \ No newline at end of file diff --git a/kotlinx-coroutines-core/jvm/test/sync/SemaphoreStressTest.kt b/kotlinx-coroutines-core/jvm/test/sync/SemaphoreStressTest.kt index 9c77990862..374a1e3d7c 100644 --- a/kotlinx-coroutines-core/jvm/test/sync/SemaphoreStressTest.kt +++ b/kotlinx-coroutines-core/jvm/test/sync/SemaphoreStressTest.kt @@ -5,7 +5,6 @@ import org.junit.Test import kotlin.test.assertEquals class SemaphoreStressTest : TestBase() { - @Test fun stressTestAsMutex() = runBlocking(Dispatchers.Default) { val n = 10_000 * stressTestMultiplier @@ -71,14 +70,14 @@ class SemaphoreStressTest : TestBase() { // Initially, we hold the permit and no one else can `acquire`, // otherwise it's a bug. assertEquals(0, semaphore.availablePermits) - var job1_entered_critical_section = false + var job1EnteredCriticalSection = false val job1 = launch(start = CoroutineStart.UNDISPATCHED) { semaphore.acquire() - job1_entered_critical_section = true + job1EnteredCriticalSection = true semaphore.release() } // check that `job1` didn't finish the call to `acquire()` - assertEquals(false, job1_entered_critical_section) + assertEquals(false, job1EnteredCriticalSection) val job2 = launch(pool) { semaphore.release() } @@ -91,5 +90,4 @@ class SemaphoreStressTest : TestBase() { } } } - } diff --git a/kotlinx-coroutines-debug/test/CoroutinesDumpTest.kt b/kotlinx-coroutines-debug/test/CoroutinesDumpTest.kt index 8507721e30..fd0279123f 100644 --- a/kotlinx-coroutines-debug/test/CoroutinesDumpTest.kt +++ b/kotlinx-coroutines-debug/test/CoroutinesDumpTest.kt @@ -115,16 +115,22 @@ class CoroutinesDumpTest : DebugTestBase() { coroutineThread!!.interrupt() val expected = - ("kotlin.coroutines.intrinsics.IntrinsicsKt__IntrinsicsJvmKt.createCoroutineUnintercepted(IntrinsicsJvm.kt:116)\n" + - "kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable(Cancellable.kt:23)\n" + - "kotlinx.coroutines.CoroutineStart.invoke(CoroutineStart.kt:109)\n" + - "kotlinx.coroutines.AbstractCoroutine.start(AbstractCoroutine.kt:160)\n" + - "kotlinx.coroutines.BuildersKt__Builders_commonKt.async(Builders.common.kt:88)\n" + - "kotlinx.coroutines.BuildersKt.async(Unknown Source)\n" + - "kotlinx.coroutines.BuildersKt__Builders_commonKt.async\$default(Builders.common.kt:81)\n" + - "kotlinx.coroutines.BuildersKt.async\$default(Unknown Source)\n" + - "kotlinx.coroutines.debug.CoroutinesDumpTest\$testCreationStackTrace\$1.invokeSuspend(CoroutinesDumpTest.kt)").trimStackTrace() - assertTrue(result.startsWith(expected)) + "kotlin.coroutines.intrinsics.IntrinsicsKt__IntrinsicsJvmKt.createCoroutineUnintercepted(IntrinsicsJvm.kt)\n" + + "kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable(Cancellable.kt)\n" + + "kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable\$default(Cancellable.kt)\n" + + "kotlinx.coroutines.CoroutineStart.invoke(CoroutineStart.kt)\n" + + "kotlinx.coroutines.AbstractCoroutine.start(AbstractCoroutine.kt)\n" + + "kotlinx.coroutines.BuildersKt__Builders_commonKt.async(Builders.common.kt)\n" + + "kotlinx.coroutines.BuildersKt.async(Unknown Source)\n" + + "kotlinx.coroutines.BuildersKt__Builders_commonKt.async\$default(Builders.common.kt)\n" + + "kotlinx.coroutines.BuildersKt.async\$default(Unknown Source)\n" + + "kotlinx.coroutines.debug.CoroutinesDumpTest\$testCreationStackTrace\$1.invokeSuspend(CoroutinesDumpTest.kt)" + if (!result.startsWith(expected)) { + println("=== Actual result") + println(result) + error("Does not start with expected lines") + } + } @Test diff --git a/kotlinx-coroutines-debug/test/DebugProbesTest.kt b/kotlinx-coroutines-debug/test/DebugProbesTest.kt index 24050e563c..3b32db3a5a 100644 --- a/kotlinx-coroutines-debug/test/DebugProbesTest.kt +++ b/kotlinx-coroutines-debug/test/DebugProbesTest.kt @@ -40,24 +40,25 @@ class DebugProbesTest : DebugTestBase() { val deferred = createDeferred() val traces = listOf( "java.util.concurrent.ExecutionException\n" + - "\tat kotlinx.coroutines.debug.DebugProbesTest\$createDeferred\$1.invokeSuspend(DebugProbesTest.kt:16)\n" + + "\tat kotlinx.coroutines.debug.DebugProbesTest\$createDeferred\$1.invokeSuspend(DebugProbesTest.kt)\n" + "\t(Coroutine boundary)\n" + "\tat kotlinx.coroutines.DeferredCoroutine.await\$suspendImpl(Builders.common.kt)\n" + - "\tat kotlinx.coroutines.debug.DebugProbesTest.oneMoreNestedMethod(DebugProbesTest.kt:71)\n" + - "\tat kotlinx.coroutines.debug.DebugProbesTest.nestedMethod(DebugProbesTest.kt:66)\n" + + "\tat kotlinx.coroutines.debug.DebugProbesTest.oneMoreNestedMethod(DebugProbesTest.kt)\n" + + "\tat kotlinx.coroutines.debug.DebugProbesTest.nestedMethod(DebugProbesTest.kt)\n" + "\t(Coroutine creation stacktrace)\n" + - "\tat kotlin.coroutines.intrinsics.IntrinsicsKt__IntrinsicsJvmKt.createCoroutineUnintercepted(IntrinsicsJvm.kt:116)\n" + - "\tat kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable(Cancellable.kt:23)\n" + - "\tat kotlinx.coroutines.CoroutineStart.invoke(CoroutineStart.kt:99)\n" + - "\tat kotlinx.coroutines.AbstractCoroutine.start(AbstractCoroutine.kt:148)\n" + - "\tat kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking(Builders.kt:45)\n" + + "\tat kotlin.coroutines.intrinsics.IntrinsicsKt__IntrinsicsJvmKt.createCoroutineUnintercepted(IntrinsicsJvm.kt)\n" + + "\tat kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable(Cancellable.kt)\n" + + "\tat kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable\$default(Cancellable.kt)\n" + + "\tat kotlinx.coroutines.CoroutineStart.invoke(CoroutineStart.kt)\n" + + "\tat kotlinx.coroutines.AbstractCoroutine.start(AbstractCoroutine.kt)\n" + + "\tat kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking(Builders.kt)\n" + "\tat kotlinx.coroutines.BuildersKt.runBlocking(Unknown Source)\n" + - "\tat kotlinx.coroutines.TestBase.runTest(TestBase.kt:138)\n" + - "\tat kotlinx.coroutines.TestBase.runTest\$default(TestBase.kt:19)\n" + - "\tat kotlinx.coroutines.debug.DebugProbesTest.testAsyncWithProbes(DebugProbesTest.kt:38)", + "\tat kotlinx.coroutines.TestBase.runTest(TestBase.kt)\n" + + "\tat kotlinx.coroutines.TestBase.runTest\$default(TestBase.kt)\n" + + "\tat kotlinx.coroutines.debug.DebugProbesTest.testAsyncWithProbes(DebugProbesTest.kt)", "Caused by: java.util.concurrent.ExecutionException\n" + - "\tat kotlinx.coroutines.debug.DebugProbesTest\$createDeferred\$1.invokeSuspend(DebugProbesTest.kt:16)\n" + - "\tat kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:32)\n") + "\tat kotlinx.coroutines.debug.DebugProbesTest\$createDeferred\$1.invokeSuspend(DebugProbesTest.kt)\n" + + "\tat kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt)\n") nestedMethod(deferred, traces) deferred.join() }