Minimize cut-and-pasted code between JS and JVM
diff --git a/core/kotlinx-coroutines-core/README.md b/core/kotlinx-coroutines-core/README.md
index ab1759e..1e1c722 100644
--- a/core/kotlinx-coroutines-core/README.md
+++ b/core/kotlinx-coroutines-core/README.md
@@ -87,6 +87,10 @@
Low-level primitives for finer-grained control of coroutines.
+# Package kotlinx.coroutines.experimental.timeunit
+
+Optional time unit support for multiplatform projects.
+
<!--- MODULE kotlinx-coroutines-core -->
<!--- INDEX kotlinx.coroutines.experimental -->
[launch]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/launch.html
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/AbstractContinuation.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/AbstractContinuation.kt
deleted file mode 100644
index f747f02..0000000
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/AbstractContinuation.kt
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental
-
-import kotlinx.atomicfu.atomic
-import kotlinx.atomicfu.loop
-import kotlin.coroutines.experimental.Continuation
-import kotlin.coroutines.experimental.intrinsics.COROUTINE_SUSPENDED
-
-private const val UNDECIDED = 0
-private const val SUSPENDED = 1
-private const val RESUMED = 2
-
-/**
- * @suppress **This is unstable API and it is subject to change.**
- */
-internal abstract class AbstractContinuation<in T>(
- public final override val delegate: Continuation<T>,
- public final override val resumeMode: Int
-) : JobSupport(true), Continuation<T>, DispatchedTask<T> {
- private val _decision = atomic(UNDECIDED)
-
- /* decision state machine
-
- +-----------+ trySuspend +-----------+
- | UNDECIDED | -------------> | SUSPENDED |
- +-----------+ +-----------+
- |
- | tryResume
- V
- +-----------+
- | RESUMED |
- +-----------+
-
- Note: both tryResume and trySuspend can be invoked at most once, first invocation wins
- */
-
- override fun takeState(): Any? = state
-
- private fun trySuspend(): Boolean {
- _decision.loop { decision ->
- when (decision) {
- UNDECIDED -> if (this._decision.compareAndSet(UNDECIDED, SUSPENDED)) return true
- RESUMED -> return false
- else -> error("Already suspended")
- }
- }
- }
-
- private fun tryResume(): Boolean {
- _decision.loop { decision ->
- when (decision) {
- UNDECIDED -> if (this._decision.compareAndSet(UNDECIDED, RESUMED)) return true
- SUSPENDED -> return false
- else -> error("Already resumed")
- }
- }
- }
-
- @PublishedApi
- internal fun getResult(): Any? {
- if (trySuspend()) return COROUTINE_SUSPENDED
- // otherwise, onCompletionInternal was already invoked & invoked tryResume, and the result is in the state
- val state = this.state
- if (state is CompletedExceptionally) throw state.exception
- return getSuccessfulResult(state)
- }
-
- internal final override fun onCompletionInternal(state: Any?, mode: Int) {
- if (tryResume()) return // completed before getResult invocation -- bail out
- // otherwise, getResult has already commenced, i.e. completed later or in other thread
- dispatch(mode)
- }
-
- override fun resume(value: T) =
- resumeImpl(value, resumeMode)
-
- override fun resumeWithException(exception: Throwable) =
- resumeImpl(CompletedExceptionally(exception), resumeMode)
-
- protected fun resumeImpl(proposedUpdate: Any?, resumeMode: Int) {
- loopOnState { state ->
- when (state) {
- is Incomplete -> {
- if (updateState(state, proposedUpdate, resumeMode)) return
- }
- is Cancelled -> {
- // Ignore resumes in cancelled coroutines, but handle exception if a different one here
- if (proposedUpdate is CompletedExceptionally && proposedUpdate.exception != state.exception)
- handleException(proposedUpdate.exception)
- return
- }
- else -> error("Already resumed, but got $proposedUpdate")
- }
- }
- }
-
- override fun handleException(exception: Throwable) {
- handleCoroutineException(context, exception)
- }
-}
\ No newline at end of file
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Annotations.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Annotations.kt
new file mode 100644
index 0000000..21e4c78
--- /dev/null
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Annotations.kt
@@ -0,0 +1,29 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kotlinx.coroutines.experimental.internalAnnotations
+
+@Suppress("ACTUAL_WITHOUT_EXPECT")
+internal actual typealias JvmName = kotlin.jvm.JvmName
+
+@Suppress("ACTUAL_WITHOUT_EXPECT")
+internal actual typealias JvmMultifileClass = kotlin.jvm.JvmMultifileClass
+
+@Suppress("ACTUAL_WITHOUT_EXPECT")
+internal actual typealias JvmField = kotlin.jvm.JvmField
+
+@Suppress("ACTUAL_WITHOUT_EXPECT")
+internal actual typealias Volatile = kotlin.jvm.Volatile
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt
index d54986c..7d20709 100644
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt
@@ -14,153 +14,13 @@
* limitations under the License.
*/
-// :todo: Remove after transition to Kotlin 1.2.30+
-@file:Suppress("ACTUAL_FUNCTION_WITH_DEFAULT_ARGUMENTS")
+@file:JvmMultifileClass
+@file:JvmName("BuildersKt")
package kotlinx.coroutines.experimental
-import kotlinx.coroutines.experimental.intrinsics.*
import java.util.concurrent.locks.*
import kotlin.coroutines.experimental.*
-import kotlin.coroutines.experimental.intrinsics.*
-
-// --------------- basic coroutine builders ---------------
-
-/**
- * Launches new coroutine without blocking current thread and returns a reference to the coroutine as a [Job].
- * The coroutine is cancelled when the resulting job is [cancelled][Job.cancel].
- *
- * The [context] for the new coroutine can be explicitly specified.
- * See [CoroutineDispatcher] for the standard context implementations that are provided by `kotlinx.coroutines`.
- * The [coroutineContext](https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.coroutines.experimental/coroutine-context.html)
- * of the parent coroutine may be used,
- * in which case the [Job] of the resulting coroutine is a child of the job of the parent coroutine.
- * The parent job may be also explicitly specified using [parent] parameter.
- *
- * If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [DefaultDispatcher] is used.
- *
- * By default, the coroutine is immediately scheduled for execution.
- * Other options can be specified via `start` parameter. See [CoroutineStart] for details.
- * An optional [start] parameter can be set to [CoroutineStart.LAZY] to start coroutine _lazily_. In this case,
- * the coroutine [Job] is created in _new_ state. It can be explicitly started with [start][Job.start] function
- * and will be started implicitly on the first invocation of [join][Job.join].
- *
- * Uncaught exceptions in this coroutine cancel parent job in the context by default
- * (unless [CoroutineExceptionHandler] is explicitly specified), which means that when `launch` is used with
- * the context of another coroutine, then any uncaught exception leads to the cancellation of parent coroutine.
- *
- * See [newCoroutineContext] for a description of debugging facilities that are available for newly created coroutine.
- *
- * @param context context of the coroutine. The default value is [DefaultDispatcher].
- * @param start coroutine start option. The default value is [CoroutineStart.DEFAULT].
- * @param parent explicitly specifies the parent job, overrides job from the [context] (if any).
- * @param onCompletion optional completion handler for the coroutine (see [Job.invokeOnCompletion]).
- * @param block the coroutine code.
- */
-public actual fun launch(
- context: CoroutineContext = DefaultDispatcher,
- start: CoroutineStart = CoroutineStart.DEFAULT,
- parent: Job? = null,
- onCompletion: CompletionHandler? = null,
- block: suspend CoroutineScope.() -> Unit
-): Job {
- val newContext = newCoroutineContext(context, parent)
- val coroutine = if (start.isLazy)
- LazyStandaloneCoroutine(newContext, block) else
- StandaloneCoroutine(newContext, active = true)
- if (onCompletion != null) coroutine.invokeOnCompletion(handler = onCompletion)
- coroutine.start(start, coroutine, block)
- return coroutine
-}
-
-/** @suppress **Deprecated**: Binary compatibility */
-@Deprecated(message = "Binary compatibility", level = DeprecationLevel.HIDDEN)
-public fun launch(
- context: CoroutineContext = DefaultDispatcher,
- start: CoroutineStart = CoroutineStart.DEFAULT,
- parent: Job? = null,
- block: suspend CoroutineScope.() -> Unit
-): Job = launch(context, start, parent, block = block)
-
-/** @suppress **Deprecated**: Binary compatibility */
-@Deprecated(message = "Binary compatibility", level = DeprecationLevel.HIDDEN)
-public fun launch(
- context: CoroutineContext = DefaultDispatcher,
- start: CoroutineStart = CoroutineStart.DEFAULT,
- block: suspend CoroutineScope.() -> Unit
-): Job =
- launch(context, start, block = block)
-
-/**
- * @suppress **Deprecated**: Use `start = CoroutineStart.XXX` parameter
- */
-@Deprecated(message = "Use `start = CoroutineStart.XXX` parameter",
- replaceWith = ReplaceWith("launch(context, if (start) CoroutineStart.DEFAULT else CoroutineStart.LAZY, block)"))
-public fun launch(context: CoroutineContext, start: Boolean, block: suspend CoroutineScope.() -> Unit): Job =
- launch(context, if (start) CoroutineStart.DEFAULT else CoroutineStart.LAZY, block = block)
-
-/**
- * Calls the specified suspending block with a given coroutine context, suspends until it completes, and returns
- * the result.
- *
- * This function immediately applies dispatcher from the new context, shifting execution of the block into the
- * different thread inside the block, and back when it completes.
- * The specified [context] is added onto the current coroutine context for the execution of the block.
- *
- * An optional `start` parameter is used only if the specified `context` uses a different [CoroutineDispatcher] than
- * a current one, otherwise it is ignored.
- * By default, the coroutine is immediately scheduled for execution and can be cancelled
- * while it is waiting to be executed and it can be cancelled while the result is scheduled
- * to be processed by the invoker context.
- * Other options can be specified via `start` parameter. See [CoroutineStart] for details.
- * A value of [CoroutineStart.LAZY] is not supported and produces [IllegalArgumentException].
- */
-public actual suspend fun <T> withContext(
- context: CoroutineContext,
- start: CoroutineStart = CoroutineStart.DEFAULT,
- block: suspend () -> T
-): T = suspendCoroutineOrReturn sc@ { cont ->
- val oldContext = cont.context
- // fast path #1 if there is no change in the actual context:
- if (context === oldContext || context is CoroutineContext.Element && oldContext[context.key] === context)
- return@sc block.startCoroutineUninterceptedOrReturn(cont)
- // compute new context
- val newContext = oldContext + context
- // fast path #2 if the result is actually the same
- if (newContext === oldContext)
- return@sc block.startCoroutineUninterceptedOrReturn(cont)
- // fast path #3 if the new dispatcher is the same as the old one.
- // `equals` is used by design (see equals implementation is wrapper context like ExecutorCoroutineDispatcher)
- if (newContext[ContinuationInterceptor] == oldContext[ContinuationInterceptor]) {
- val newContinuation = RunContinuationDirect(newContext, cont)
- return@sc block.startCoroutineUninterceptedOrReturn(newContinuation)
- }
- // slowest path otherwise -- use new interceptor, sync to its result via a full-blown instance of RunCompletion
- require(!start.isLazy) { "$start start is not supported" }
- val completion = RunCompletion(
- context = newContext,
- delegate = cont,
- resumeMode = if (start == CoroutineStart.ATOMIC) MODE_ATOMIC_DEFAULT else MODE_CANCELLABLE)
- completion.initParentJobInternal(newContext[Job]) // attach to job
- @Suppress("DEPRECATION")
- start(block, completion)
- completion.getResult()
-}
-
-/** @suppress **Deprecated**: Renamed to [withContext]. */
-@Deprecated(message = "Renamed to `withContext`", level=DeprecationLevel.WARNING,
- replaceWith = ReplaceWith("withContext(context, start, block)"))
-public suspend fun <T> run(
- context: CoroutineContext,
- start: CoroutineStart = CoroutineStart.DEFAULT,
- block: suspend () -> T
-): T =
- withContext(context, start, block)
-
-/** @suppress **Deprecated** */
-@Deprecated(message = "It is here for binary compatibility only", level=DeprecationLevel.HIDDEN)
-public suspend fun <T> run(context: CoroutineContext, block: suspend () -> T): T =
- withContext(context, start = CoroutineStart.ATOMIC, block = block)
/**
* Runs new coroutine and **blocks** current thread _interruptibly_ until its completion.
@@ -198,41 +58,6 @@
return coroutine.joinBlocking()
}
-// --------------- implementation ---------------
-
-private open class StandaloneCoroutine(
- private val parentContext: CoroutineContext,
- active: Boolean
-) : AbstractCoroutine<Unit>(parentContext, active) {
- override fun hasOnFinishingHandler(update: Any?) = update is CompletedExceptionally
- override fun onFinishingInternal(update: Any?) {
- // note the use of the parent's job context below!
- if (update is CompletedExceptionally) handleCoroutineException(parentContext, update.exception)
- }
-}
-
-private class LazyStandaloneCoroutine(
- parentContext: CoroutineContext,
- private val block: suspend CoroutineScope.() -> Unit
-) : StandaloneCoroutine(parentContext, active = false) {
- override fun onStart() {
- block.startCoroutineCancellable(this, this)
- }
-}
-
-private class RunContinuationDirect<in T>(
- override val context: CoroutineContext,
- continuation: Continuation<T>
-) : Continuation<T> by continuation
-
-
-@Suppress("UNCHECKED_CAST")
-private class RunCompletion<in T>(
- override val context: CoroutineContext,
- delegate: Continuation<T>,
- resumeMode: Int
-) : AbstractContinuation<T>(delegate, resumeMode)
-
private class BlockingCoroutine<T>(
parentContext: CoroutineContext,
private val blockedThread: Thread,
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CancellableContinuation.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CancellableContinuation.kt
deleted file mode 100644
index bba910f..0000000
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CancellableContinuation.kt
+++ /dev/null
@@ -1,295 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-// :todo: Remove after transition to Kotlin 1.2.30+
-@file:Suppress("ACTUAL_FUNCTION_WITH_DEFAULT_ARGUMENTS")
-
-package kotlinx.coroutines.experimental
-
-import kotlinx.coroutines.experimental.internal.LockFreeLinkedListNode
-import kotlin.coroutines.experimental.Continuation
-import kotlin.coroutines.experimental.CoroutineContext
-import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
-import kotlin.coroutines.experimental.suspendCoroutine
-
-// --------------- cancellable continuations ---------------
-
-/**
- * Cancellable continuation. Its job is _completed_ when it is resumed or cancelled.
- * When [cancel] function is explicitly invoked, this continuation immediately resumes with [CancellationException] or
- * with the specified cancel cause.
- *
- * Cancellable continuation has three states (as subset of [Job] states):
- *
- * | **State** | [isActive] | [isCompleted] | [isCancelled] |
- * | ----------------------------------- | ---------- | ------------- | ------------- |
- * | _Active_ (initial state) | `true` | `false` | `false` |
- * | _Resumed_ (final _completed_ state) | `false` | `true` | `false` |
- * | _Canceled_ (final _completed_ state)| `false` | `true` | `true` |
- *
- * Invocation of [cancel] transitions this continuation from _active_ to _cancelled_ state, while
- * invocation of [resume] or [resumeWithException] transitions it from _active_ to _resumed_ state.
- *
- * A [cancelled][isCancelled] continuation implies that it is [completed][isCompleted].
- *
- * Invocation of [resume] or [resumeWithException] in _resumed_ state produces [IllegalStateException]
- * but is ignored in _cancelled_ state.
- *
- * ```
- * +-----------+ resume +---------+
- * | Active | ----------> | Resumed |
- * +-----------+ +---------+
- * |
- * | cancel
- * V
- * +-----------+
- * | Cancelled |
- * +-----------+
- *
- * ```
- */
-public actual interface CancellableContinuation<in T> : Continuation<T>, Job {
- /**
- * Returns `true` when this continuation is active -- it has not completed or cancelled yet.
- */
- public actual override val isActive: Boolean
-
- /**
- * Returns `true` when this continuation has completed for any reason. A continuation
- * that was cancelled is also considered complete.
- */
- public actual override val isCompleted: Boolean
-
- /**
- * Returns `true` if this continuation was [cancelled][cancel].
- *
- * It implies that [isActive] is `false` and [isCompleted] is `true`.
- */
- public actual override val isCancelled: Boolean
-
- /**
- * Tries to resume this continuation with a given value and returns non-null object token if it was successful,
- * or `null` otherwise (it was already resumed or cancelled). When non-null object was returned,
- * [completeResume] must be invoked with it.
- *
- * When [idempotent] is not `null`, this function performs _idempotent_ operation, so that
- * further invocations with the same non-null reference produce the same result.
- *
- * @suppress **This is unstable API and it is subject to change.**
- */
- public actual fun tryResume(value: T, idempotent: Any? = null): Any?
-
- /**
- * Tries to resume this continuation with a given exception and returns non-null object token if it was successful,
- * or `null` otherwise (it was already resumed or cancelled). When non-null object was returned,
- * [completeResume] must be invoked with it.
- *
- * @suppress **This is unstable API and it is subject to change.**
- */
- public fun tryResumeWithException(exception: Throwable): Any?
-
- /**
- * Completes the execution of [tryResume] or [tryResumeWithException] on its non-null result.
- *
- * @suppress **This is unstable API and it is subject to change.**
- */
- public actual fun completeResume(token: Any)
-
- /**
- * Makes this continuation cancellable. Use it with `holdCancellability` optional parameter to
- * [suspendCancellableCoroutine] function. It throws [IllegalStateException] if invoked more than once.
- */
- public actual fun initCancellability()
-
- /**
- * Cancels this continuation with an optional cancellation [cause]. The result is `true` if this continuation was
- * cancelled as a result of this invocation and `false` otherwise.
- */
- @Suppress("DEFAULT_VALUE_NOT_ALLOWED_IN_OVERRIDE")
- public actual override fun cancel(cause: Throwable? = null): Boolean
-
- /**
- * Registers handler that is **synchronously** invoked once on completion of this continuation.
- * When continuation is already complete, then the handler is immediately invoked
- * with continuation's exception or `null`. Otherwise, handler will be invoked once when this
- * continuation is complete.
- *
- * The resulting [DisposableHandle] can be used to [dispose][DisposableHandle.dispose] the
- * registration of this handler and release its memory if its invocation is no longer needed.
- * There is no need to dispose the handler after completion of this continuation. The references to
- * all the handlers are released when this continuation completes.
- *
- * Installed [handler] should not throw any exceptions. If it does, they will get caught,
- * wrapped into [CompletionHandlerException], and rethrown, potentially causing crash of unrelated code.
- */
- public actual override fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle
-
- /**
- * Resumes this continuation with a given [value] in the invoker thread without going though
- * [dispatch][CoroutineDispatcher.dispatch] function of the [CoroutineDispatcher] in the [context].
- * This function is designed to be used only by the [CoroutineDispatcher] implementations themselves.
- * **It should not be used in general code**.
- */
- public actual fun CoroutineDispatcher.resumeUndispatched(value: T)
-
- /**
- * Resumes this continuation with a given [exception] in the invoker thread without going though
- * [dispatch][CoroutineDispatcher.dispatch] function of the [CoroutineDispatcher] in the [context].
- * This function is designed to be used only by the [CoroutineDispatcher] implementations themselves.
- * **It should not be used in general code**.
- */
- public actual fun CoroutineDispatcher.resumeUndispatchedWithException(exception: Throwable)
-}
-
-/**
- * Suspends coroutine similar to [suspendCoroutine], but provide an implementation of [CancellableContinuation] to
- * the [block]. This function throws [CancellationException] if the coroutine is cancelled or completed while suspended.
- *
- * If [holdCancellability] optional parameter is `true`, then the coroutine is suspended, but it is not
- * cancellable until [CancellableContinuation.initCancellability] is invoked.
- *
- * See [suspendAtomicCancellableCoroutine] for suspending functions that need *atomic cancellation*.
- */
-public actual inline suspend fun <T> suspendCancellableCoroutine(
- holdCancellability: Boolean = false,
- crossinline block: (CancellableContinuation<T>) -> Unit
-): T =
- suspendCoroutineOrReturn { cont ->
- val cancellable = CancellableContinuationImpl(cont, resumeMode = MODE_CANCELLABLE)
- if (!holdCancellability) cancellable.initCancellability()
- block(cancellable)
- cancellable.getResult()
-}
-
-/**
- * Suspends coroutine similar to [suspendCancellableCoroutine], but with *atomic cancellation*.
- *
- * When suspended function throws [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.
- */
-public actual inline suspend fun <T> suspendAtomicCancellableCoroutine(
- holdCancellability: Boolean = false,
- crossinline block: (CancellableContinuation<T>) -> Unit
-): T =
- suspendCoroutineOrReturn { cont ->
- val cancellable = CancellableContinuationImpl(cont, resumeMode = MODE_ATOMIC_DEFAULT)
- if (!holdCancellability) cancellable.initCancellability()
- block(cancellable)
- cancellable.getResult()
- }
-
-/**
- * Removes a given node on cancellation.
- * @suppress **This is unstable API and it is subject to change.**
- */
-public fun CancellableContinuation<*>.removeOnCancel(node: LockFreeLinkedListNode): DisposableHandle =
- invokeOnCompletion(handler = RemoveOnCancel(this, node))
-
-// --------------- implementation details ---------------
-
-private class RemoveOnCancel(
- cont: CancellableContinuation<*>,
- val node: LockFreeLinkedListNode
-) : JobNode<CancellableContinuation<*>>(cont) {
- override fun invoke(reason: Throwable?) {
- if (job.isCancelled)
- node.remove()
- }
- override fun toString() = "RemoveOnCancel[$node]"
-}
-
-@PublishedApi
-internal class CancellableContinuationImpl<in T>(
- delegate: Continuation<T>,
- resumeMode: Int
-) : AbstractContinuation<T>(delegate, resumeMode), CancellableContinuation<T>, Runnable {
- @Volatile // just in case -- we don't want an extra data race, even benign one
- private var _context: CoroutineContext? = null // created on first need
-
- public override val context: CoroutineContext
- get() = _context ?: (delegate.context + this).also { _context = it }
-
- override fun initCancellability() {
- initParentJobInternal(delegate.context[Job])
- }
-
- override val onCancelMode: Int get() = ON_CANCEL_MAKE_CANCELLED
-
- override fun tryResume(value: T, idempotent: Any?): Any? {
- while (true) { // lock-free loop on state
- val state = this.state // atomic read
- when (state) {
- is Incomplete -> {
- val update: Any? = if (idempotent == null) value else
- CompletedIdempotentResult(idempotent, value, state)
- if (tryUpdateState(state, update)) return state
- }
- is CompletedIdempotentResult -> {
- if (state.idempotentResume === idempotent) {
- check(state.result === value) { "Non-idempotent resume" }
- return state.token
- } else
- return null
- }
- else -> return null // cannot resume -- not active anymore
- }
- }
- }
-
- override fun tryResumeWithException(exception: Throwable): Any? {
- while (true) { // lock-free loop on state
- val state = this.state // atomic read
- when (state) {
- is Incomplete -> {
- if (tryUpdateState(state, CompletedExceptionally(exception))) return state
- }
- else -> return null // cannot resume -- not active anymore
- }
- }
- }
-
- override fun completeResume(token: Any) {
- completeUpdateState(token as Incomplete, state, resumeMode)
- }
-
- override fun CoroutineDispatcher.resumeUndispatched(value: T) {
- val dc = delegate as? DispatchedContinuation
- resumeImpl(value, if (dc?.dispatcher === this) MODE_UNDISPATCHED else resumeMode)
- }
-
- override fun CoroutineDispatcher.resumeUndispatchedWithException(exception: Throwable) {
- val dc = delegate as? DispatchedContinuation
- resumeImpl(CompletedExceptionally(exception), if (dc?.dispatcher === this) MODE_UNDISPATCHED else resumeMode)
- }
-
- @Suppress("UNCHECKED_CAST")
- override fun <T> getSuccessfulResult(state: Any?): T =
- if (state is CompletedIdempotentResult) state.result as T else state as T
-
- override fun nameString(): String =
- "CancellableContinuation(${delegate.toDebugString()})"
-}
-
-private class CompletedIdempotentResult(
- @JvmField val idempotentResume: Any?,
- @JvmField val result: Any?,
- @JvmField val token: JobSupport.Incomplete
-) {
- override fun toString(): String = "CompletedIdempotentResult[$result]"
-}
-
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CommonPool.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CommonPool.kt
index a73ad00..10081ef 100644
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CommonPool.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CommonPool.kt
@@ -16,6 +16,7 @@
package kotlinx.coroutines.experimental
+import kotlinx.coroutines.experimental.timeunit.TimeUnit
import java.util.concurrent.*
import java.util.concurrent.atomic.AtomicInteger
import kotlin.coroutines.experimental.CoroutineContext
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CompletableDeferred.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CompletableDeferred.kt
deleted file mode 100644
index af61dd9..0000000
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CompletableDeferred.kt
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-// :todo: Remove after transition to Kotlin 1.2.30+
-@file:Suppress("ACTUAL_FUNCTION_WITH_DEFAULT_ARGUMENTS")
-
-package kotlinx.coroutines.experimental
-
-import kotlinx.coroutines.experimental.selects.*
-
-/**
- * A [Deferred] that can be completed via public functions
- * [complete], [completeExceptionally], and [cancel].
- *
- * Completion functions return `false` when this deferred value is already complete or completing.
- *
- * An instance of completable deferred can be created by `CompletableDeferred()` function in _active_ state.
- *
- * All functions on this interface and on all interfaces derived from it are **thread-safe** and can
- * be safely invoked from concurrent coroutines without external synchronization.
- */
-public actual interface CompletableDeferred<T> : Deferred<T> {
- /**
- * Completes this deferred value with a given [value]. The result is `true` if this deferred was
- * completed as a result of this invocation and `false` otherwise (if it was already completed).
- *
- * Repeated invocations of this function have no effect and always produce `false`.
- */
- public actual fun complete(value: T): Boolean
-
- /**
- * Completes this deferred value exceptionally with a given [exception]. The result is `true` if this deferred was
- * completed as a result of this invocation and `false` otherwise (if it was already completed).
- *
- * Repeated invocations of this function have no effect and always produce `false`.
- */
- public actual fun completeExceptionally(exception: Throwable): Boolean
-}
-
-/**
- * Creates a [CompletableDeferred] in an _active_ state.
- * It is optionally a child of a [parent] job.
- */
-@Suppress("FunctionName")
-public actual fun <T> CompletableDeferred(parent: Job? = null): CompletableDeferred<T> = CompletableDeferredImpl(parent)
-
-/** @suppress **Deprecated:** Binary compatibility only */
-@Deprecated(message = "Binary compatibility only", level = DeprecationLevel.HIDDEN)
-@Suppress("FunctionName")
-public fun <T> CompletableDeferred(): CompletableDeferred<T> = CompletableDeferredImpl(null)
-
-/**
- * Creates an already _completed_ [CompletableDeferred] with a given [value].
- */
-@Suppress("FunctionName")
-public actual fun <T> CompletableDeferred(value: T): CompletableDeferred<T> = CompletableDeferredImpl<T>(null).apply { complete(value) }
-
-/**
- * Concrete implementation of [CompletableDeferred].
- */
-@Suppress("UNCHECKED_CAST")
-private class CompletableDeferredImpl<T>(
- parent: Job?
-) : JobSupport(true), CompletableDeferred<T>, SelectClause1<T> {
- init { initParentJobInternal(parent) }
- override val onCancelMode: Int get() = ON_CANCEL_MAKE_COMPLETING
- override fun getCompleted(): T = getCompletedInternal() as T
- override suspend fun await(): T = awaitInternal() as T
- override val onAwait: SelectClause1<T> get() = this
- override fun <R> registerSelectClause1(select: SelectInstance<R>, block: suspend (T) -> R) =
- registerSelectClause1Internal(select, block)
-
- override fun complete(value: T): Boolean =
- makeCompleting(value)
- override fun completeExceptionally(exception: Throwable): Boolean =
- makeCompleting(CompletedExceptionally(exception))
-}
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/JobTest.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CompletionHandler.kt
similarity index 60%
copy from core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/JobTest.kt
copy to core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CompletionHandler.kt
index 06f7c68..0f5cd77 100644
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/JobTest.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CompletionHandler.kt
@@ -16,14 +16,12 @@
package kotlinx.coroutines.experimental
-import kotlin.test.*
+import kotlinx.coroutines.experimental.internal.*
-class JobTest : TestBase() {
- @Test
- fun testMemoryRelease() {
- val job = Job()
- val n = 10_000_000 * stressTestMultiplier
- var fireCount = 0
- for (i in 0 until n) job.invokeOnCompletion { fireCount++ }.dispose()
- }
-}
\ No newline at end of file
+internal actual abstract class CompletionHandlerNode actual constructor() : LockFreeLinkedListNode(), CompletionHandler {
+ actual inline val asHandler: CompletionHandler get() = this
+ actual abstract override fun invoke(cause: Throwable?)
+}
+
+@Suppress("NOTHING_TO_INLINE")
+internal actual inline fun CompletionHandler.invokeIt(cause: Throwable?) = invoke(cause)
\ No newline at end of file
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineContext.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineContext.kt
index 832e1f0..774ebe7 100644
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineContext.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineContext.kt
@@ -42,30 +42,6 @@
}
/**
- * A coroutine dispatcher that is not confined to any specific thread.
- * It executes initial continuation of the coroutine _right here_ in the current call-frame
- * and let the coroutine resume in whatever thread that is used by the corresponding suspending function, without
- * mandating any specific threading policy.
- *
- * Note, that if you need your coroutine to be confined to a particular thread or a thread-pool after resumption,
- * but still want to execute it in the current call-frame until its first suspension, then you can use
- * an optional [CoroutineStart] parameter in coroutine builders like [launch] and [async] setting it to the
- * the value of [CoroutineStart.UNDISPATCHED].
- */
-public actual object Unconfined : CoroutineDispatcher() {
- actual override fun isDispatchNeeded(context: CoroutineContext): Boolean = false
- actual override fun dispatch(context: CoroutineContext, block: Runnable) { throw UnsupportedOperationException() }
- override fun toString(): String = "Unconfined"
-}
-
-/**
- * @suppress **Deprecated**: `Here` was renamed to `Unconfined`.
- */
-@Deprecated(message = "`Here` was renamed to `Unconfined`",
- replaceWith = ReplaceWith(expression = "Unconfined"))
-public typealias Here = Unconfined
-
-/**
* This is the default [CoroutineDispatcher] that is used by all standard builders like
* [launch], [async], etc if no dispatcher nor any other [ContinuationInterceptor] is specified in their context.
*
@@ -94,7 +70,7 @@
* The string "coroutine" is used as a default name.
*/
@JvmOverloads // for binary compatibility with newCoroutineContext(context: CoroutineContext) version
-public fun newCoroutineContext(context: CoroutineContext, parent: Job? = null): CoroutineContext {
+public actual fun newCoroutineContext(context: CoroutineContext, parent: Job? = null): CoroutineContext {
val debug = if (DEBUG) context + CoroutineId(COROUTINE_ID.incrementAndGet()) else context
val wp = if (parent == null) debug else debug + parent
return if (context !== DefaultDispatcher && context[ContinuationInterceptor] == null)
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineDispatcher.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineDispatcher.kt
deleted file mode 100644
index feb4b0d..0000000
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineDispatcher.kt
+++ /dev/null
@@ -1,108 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental
-
-import kotlin.coroutines.experimental.*
-
-/**
- * Base class that shall be extended by all coroutine dispatcher implementations.
- *
- * The following standard implementations are provided by `kotlinx.coroutines`:
- * * [Unconfined] -- starts coroutine execution in the current call-frame until the first suspension.
- * On first suspension the coroutine builder function returns.
- * The coroutine will resume in whatever thread that is used by the
- * corresponding suspending function, without confining it to any specific thread or pool.
- * This in an appropriate choice for IO-intensive coroutines that do not consume CPU resources.
- * * [DefaultDispatcher] -- is used by all standard builder if no dispatcher nor any other [ContinuationInterceptor]
- * is specified in their context. It is currently equal to [CommonPool] (subject to change).
- * * [CommonPool] -- immediately returns from the coroutine builder and schedules coroutine execution to
- * a common pool of shared background threads.
- * This is an appropriate choice for compute-intensive coroutines that consume a lot of CPU resources.
- * * Private thread pools can be created with [newSingleThreadContext] and [newFixedThreadPoolContext].
- * * An arbitrary [Executor][java.util.concurrent.Executor] can be converted to dispatcher with [asCoroutineDispatcher] extension function.
- *
- * This class ensures that debugging facilities in [newCoroutineContext] function work properly.
- */
-public actual abstract class CoroutineDispatcher actual constructor() :
- AbstractCoroutineContextElement(ContinuationInterceptor), ContinuationInterceptor {
- /**
- * Returns `true` if execution shall be dispatched onto another thread.
- * The default behaviour for most dispatchers is to return `true`.
- *
- * UI dispatchers _should not_ override `isDispatchNeeded`, but leave a default implementation that
- * returns `true`. To understand the rationale beyond this recommendation, consider the following code:
- *
- * ```kotlin
- * fun asyncUpdateUI() = async(MainThread) {
- * // do something here that updates something in UI
- * }
- * ```
- *
- * When you invoke `asyncUpdateUI` in some background thread, it immediately continues to the next
- * line, while UI update happens asynchronously in the UI thread. However, if you invoke
- * it in the UI thread itself, it updates UI _synchronously_ if your `isDispatchNeeded` is
- * overridden with a thread check. Checking if we are already in the UI thread seems more
- * efficient (and it might indeed save a few CPU cycles), but this subtle and context-sensitive
- * difference in behavior makes the resulting async code harder to debug.
- *
- * Basically, the choice here is between "JS-style" asynchronous approach (async actions
- * are always postponed to be executed later in the even dispatch thread) and "C#-style" approach
- * (async actions are executed in the invoker thread until the first suspension point).
- * While, C# approach seems to be more efficient, it ends up with recommendations like
- * "use `yield` if you need to ....". This is error-prone. JS-style approach is more consistent
- * and does not require programmers to think about whether they need to yield or not.
- *
- * However, coroutine builders like [launch] and [async] accept an optional [CoroutineStart]
- * parameter that allows one to optionally choose C#-style [CoroutineStart.UNDISPATCHED] behaviour
- * whenever it is needed for efficiency.
- */
- public actual open fun isDispatchNeeded(context: CoroutineContext): Boolean = true
-
- /**
- * Dispatches execution of a runnable [block] onto another thread in the given [context].
- */
- public actual abstract fun dispatch(context: CoroutineContext, block: Runnable)
-
- /**
- * Returns continuation that wraps the original [continuation], thus intercepting all resumptions.
- */
- public actual override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> =
- DispatchedContinuation(this, continuation)
-
- /**
- * @suppress **Error**: Operator '+' on two CoroutineDispatcher objects is meaningless.
- * CoroutineDispatcher is a coroutine context element and `+` is a set-sum operator for coroutine contexts.
- * The dispatcher to the right of `+` just replaces the dispatcher the left of `+`.
- */
- @Suppress("DeprecatedCallableAddReplaceWith")
- @Deprecated(
- message = "Operator '+' on two CoroutineDispatcher objects is meaningless. " +
- "CoroutineDispatcher is a coroutine context element and `+` is a set-sum operator for coroutine contexts. " +
- "The dispatcher to the right of `+` just replaces the dispatcher the left of `+`.",
- level = DeprecationLevel.ERROR
- )
- public operator fun plus(other: CoroutineDispatcher) = other
-
- // for nicer debugging
- override fun toString(): String =
- "${this::class.java.simpleName}@${Integer.toHexString(System.identityHashCode(this))}"
-}
-
-/**
- * A runnable task for [CoroutineDispatcher.dispatch].
- */
-public actual typealias Runnable = java.lang.Runnable
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineExceptionHandler.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineExceptionHandler.kt
deleted file mode 100644
index 2bb0cc7..0000000
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineExceptionHandler.kt
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental
-
-import java.util.*
-import kotlin.coroutines.experimental.AbstractCoroutineContextElement
-import kotlin.coroutines.experimental.CoroutineContext
-
-/**
- * Helper function for coroutine builder implementations to handle uncaught exception in coroutines.
- *
- * It tries to handle uncaught exception in the following way:
- * * If there is [CoroutineExceptionHandler] in the context, then it is used.
- * * Otherwise, if exception is [CancellationException] then it is ignored
- * (because that is the supposed mechanism to cancel the running coroutine)
- * * Otherwise:
- * * if there is a [Job] in the context, then [Job.cancel] is invoked;
- * * all instances of [CoroutineExceptionHandler] found via [ServiceLoader] are invoked;
- * * current thread's [Thread.uncaughtExceptionHandler] is invoked.
- */
-public actual fun handleCoroutineException(context: CoroutineContext, exception: Throwable) {
- // if exception handling fails, make sure the original exception is not lost
- try {
- context[CoroutineExceptionHandler]?.let {
- it.handleException(context, exception)
- return
- }
- // ignore CancellationException (they are normal means to terminate a coroutine)
- if (exception is CancellationException) return
- // try cancel job in the context
- context[Job]?.cancel(exception)
- // use additional extension handlers
- ServiceLoader.load(CoroutineExceptionHandler::class.java).forEach { handler ->
- handler.handleException(context, exception)
- }
- // use thread's handler
- val currentThread = Thread.currentThread()
- currentThread.uncaughtExceptionHandler.uncaughtException(currentThread, exception)
- } catch (handlerException: Throwable) {
- // simply rethrow if handler threw the original exception
- if (handlerException === exception) throw exception
- // handler itself crashed for some other reason -- that is bad -- keep both
- throw RuntimeException("Exception while trying to handle coroutine exception", exception).apply {
- addSuppressed(handlerException)
- }
- }
-}
-
-/**
- * An optional element on the coroutine context to handle uncaught exceptions.
- *
- * By default, when no handler is installed, uncaught exception are handled in the following way:
- * * If exception is [CancellationException] then it is ignored
- * (because that is the supposed mechanism to cancel the running coroutine)
- * * Otherwise:
- * * if there is a [Job] in the context, then [Job.cancel] is invoked;
- * * all instances of [CoroutineExceptionHandler] found via [ServiceLoader] are invoked;
- * * and current thread's [Thread.uncaughtExceptionHandler] is invoked.
- *
- * See [handleCoroutineException].
- */
-public actual interface CoroutineExceptionHandler : CoroutineContext.Element {
- /**
- * Key for [CoroutineExceptionHandler] instance in the coroutine context.
- */
- public actual companion object Key : CoroutineContext.Key<CoroutineExceptionHandler> {
- /**
- * Creates new [CoroutineExceptionHandler] instance.
- * @param handler a function which handles exception thrown by a coroutine
- * @suppress **Deprecated**
- */
- @Deprecated("Replaced with top-level function", level = DeprecationLevel.HIDDEN)
- public operator inline fun invoke(crossinline handler: (CoroutineContext, Throwable) -> Unit): CoroutineExceptionHandler =
- CoroutineExceptionHandler(handler)
- }
-
- /**
- * Handles uncaught [exception] in the given [context]. It is invoked
- * if coroutine has an uncaught exception. See [handleCoroutineException].
- */
- public actual fun handleException(context: CoroutineContext, exception: Throwable)
-}
-
-/**
- * Creates new [CoroutineExceptionHandler] instance.
- * @param handler a function which handles exception thrown by a coroutine
- */
-@Suppress("FunctionName")
-public actual inline fun CoroutineExceptionHandler(crossinline handler: (CoroutineContext, Throwable) -> Unit): CoroutineExceptionHandler =
- object: AbstractCoroutineContextElement(CoroutineExceptionHandler), CoroutineExceptionHandler {
- override fun handleException(context: CoroutineContext, exception: Throwable) =
- handler.invoke(context, exception)
- }
\ No newline at end of file
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineExceptionHandlerImpl.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineExceptionHandlerImpl.kt
new file mode 100644
index 0000000..04246d3
--- /dev/null
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineExceptionHandlerImpl.kt
@@ -0,0 +1,31 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kotlinx.coroutines.experimental
+
+import java.util.*
+import kotlin.coroutines.experimental.AbstractCoroutineContextElement
+import kotlin.coroutines.experimental.CoroutineContext
+
+internal actual fun handleCoroutineExceptionImpl(context: CoroutineContext, exception: Throwable) {
+ // use additional extension handlers
+ ServiceLoader.load(CoroutineExceptionHandler::class.java).forEach { handler ->
+ handler.handleException(context, exception)
+ }
+ // use thread's handler
+ val currentThread = Thread.currentThread()
+ currentThread.uncaughtExceptionHandler.uncaughtException(currentThread, exception)
+}
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Debug.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Debug.kt
index 6930f33..a86d98f 100644
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Debug.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Debug.kt
@@ -20,7 +20,7 @@
// internal debugging tools
-internal val Any.hexAddress: String
+internal actual val Any.hexAddress: String
get() = Integer.toHexString(System.identityHashCode(this))
internal fun Any?.toSafeString(): String =
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/DefaultExecutor.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/DefaultExecutor.kt
index d2aab07..7d7d67d 100644
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/DefaultExecutor.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/DefaultExecutor.kt
@@ -16,7 +16,9 @@
package kotlinx.coroutines.experimental
-import java.util.concurrent.*
+import kotlinx.coroutines.experimental.timeunit.*
+
+internal actual val DefaultDelay: Delay = DefaultExecutor
@Suppress("PLATFORM_CLASS_MAPPED_TO_KOTLIN")
internal object DefaultExecutor : EventLoopBase(), Runnable {
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Deferred.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Deferred.kt
deleted file mode 100644
index ddc8e0e..0000000
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Deferred.kt
+++ /dev/null
@@ -1,235 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-// :todo: Remove after transition to Kotlin 1.2.30+
-@file:Suppress("ACTUAL_FUNCTION_WITH_DEFAULT_ARGUMENTS")
-
-package kotlinx.coroutines.experimental
-
-import kotlinx.coroutines.experimental.intrinsics.*
-import kotlinx.coroutines.experimental.selects.*
-import kotlin.coroutines.experimental.*
-
-/**
- * Deferred value is a non-blocking cancellable future.
- *
- * It is created with [async] coroutine builder or via constructor of [CompletableDeferred] class.
- * It is in [active][isActive] state while the value is being computed.
- *
- * Deferred value has the following states:
- *
- * | **State** | [isActive] | [isCompleted] | [isCompletedExceptionally] | [isCancelled] |
- * | --------------------------------------- | ---------- | ------------- | -------------------------- | ------------- |
- * | _New_ (optional initial state) | `false` | `false` | `false` | `false` |
- * | _Active_ (default initial state) | `true` | `false` | `false` | `false` |
- * | _Completing_ (optional transient state) | `true` | `false` | `false` | `false` |
- * | _Cancelling_ (optional transient state) | `false` | `false` | `false` | `true` |
- * | _Cancelled_ (final state) | `false` | `true` | `true` | `true` |
- * | _Resolved_ (final state) | `false` | `true` | `false` | `false` |
- * | _Failed_ (final state) | `false` | `true` | `true` | `false` |
- *
- * Usually, a deferred value is created in _active_ state (it is created and started).
- * However, [async] coroutine builder has an optional `start` parameter that creates a deferred value in _new_ state
- * when this parameter is set to [CoroutineStart.LAZY].
- * Such a deferred can be be made _active_ by invoking [start], [join], or [await].
- *
- * A deferred can be _cancelled_ at any time with [cancel] function that forces it to transition to
- * _cancelling_ state immediately. Deferred that is not backed by a coroutine (see [CompletableDeferred]) and does not have
- * [children] becomes _cancelled_ on [cancel] immediately.
- * Otherwise, deferred becomes _cancelled_ when it finishes executing its code and
- * when all its children [complete][isCompleted].
- *
- * ```
- * wait children
- * +-----+ start +--------+ complete +-------------+ finish +-----------+
- * | New | ---------------> | Active | ----------> | Completing | ---+-> | Resolved |
- * +-----+ +--------+ +-------------+ | |(completed)|
- * | | | | +-----------+
- * | cancel | cancel | cancel |
- * V V | | +-----------+
- * +-----------+ finish +------------+ | +-> | Failed |
- * | Cancelled | <--------- | Cancelling | <---------------+ |(completed)|
- * |(completed)| +------------+ +-----------+
- * +-----------+
- * ```
- *
- * A deferred value is a [Job]. A job in the
- * [coroutineContext](https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.coroutines.experimental/coroutine-context.html)
- * of [async] builder represents the coroutine itself.
- * A deferred value is active while the coroutine is working and cancellation aborts the coroutine when
- * the coroutine is suspended on a _cancellable_ suspension point by throwing [CancellationException]
- * or the cancellation cause inside the coroutine.
- *
- * A deferred value can have a _parent_ job. A deferred value with a parent is cancelled when its parent is
- * cancelled or completes. Parent waits for all its [children] to complete in _completing_ or
- * _cancelling_ state. _Completing_ state is purely internal. For an outside observer a _completing_
- * deferred is still active, while internally it is waiting for its children.
- *
- * All functions on this interface and on all interfaces derived from it are **thread-safe** and can
- * be safely invoked from concurrent coroutines without external synchronization.
- */
-public actual interface Deferred<out T> : Job {
- /**
- * Returns `true` if computation of this deferred value has _completed exceptionally_ -- it had
- * either _failed_ with exception during computation or was [cancelled][cancel].
- *
- * It implies that [isActive] is `false` and [isCompleted] is `true`.
- */
- public actual val isCompletedExceptionally: Boolean
-
- /**
- * Awaits for completion of this value without blocking a thread and resumes when deferred computation is complete,
- * returning the resulting value or throwing the corresponding exception if the deferred had completed exceptionally.
- *
- * This suspending function is cancellable.
- * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
- * immediately resumes with [CancellationException].
- *
- * This function can be used in [select] invocation with [onAwait] clause.
- * Use [isCompleted] to check for completion of this deferred value without waiting.
- */
- public actual suspend fun await(): T
-
- /**
- * Clause for [select] expression of [await] suspending function that selects with the deferred value when it is
- * resolved. The [select] invocation fails if the deferred value completes exceptionally (either fails or
- * it cancelled).
- */
- public val onAwait: SelectClause1<T>
-
- /**
- * Returns *completed* result or throws [IllegalStateException] if this deferred value has not
- * [completed][isCompleted] yet. It throws the corresponding exception if this deferred has
- * [completed exceptionally][isCompletedExceptionally].
- *
- * This function is designed to be used from [invokeOnCompletion] handlers, when there is an absolute certainty that
- * the value is already complete. See also [getCompletionExceptionOrNull].
- */
- public actual fun getCompleted(): T
-
- /**
- * Returns *completion exception* result if this deferred [completed exceptionally][isCompletedExceptionally],
- * `null` if it is completed normally, or throws [IllegalStateException] if this deferred value has not
- * [completed][isCompleted] yet.
- *
- * This function is designed to be used from [invokeOnCompletion] handlers, when there is an absolute certainty that
- * the value is already complete. See also [getCompleted].
- */
- public actual fun getCompletionExceptionOrNull(): Throwable?
-
- /**
- * @suppress **Deprecated**: Use `isActive`.
- */
- @Deprecated(message = "Use `isActive`", replaceWith = ReplaceWith("isActive"))
- public val isComputing: Boolean get() = isActive
-}
-
-/**
- * Creates new coroutine and returns its future result as an implementation of [Deferred].
- *
- * The running coroutine is cancelled when the resulting object is [cancelled][Job.cancel].
- *
- * The [context] for the new coroutine can be explicitly specified.
- * See [CoroutineDispatcher] for the standard context implementations that are provided by `kotlinx.coroutines`.
- * The [coroutineContext](https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.coroutines.experimental/coroutine-context.html)
- * of the parent coroutine may be used,
- * in which case the [Job] of the resulting coroutine is a child of the job of the parent coroutine.
- * The parent job may be also explicitly specified using [parent] parameter.
- *
- * If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [DefaultDispatcher] is used.
- *
- * By default, the coroutine is immediately scheduled for execution.
- * Other options can be specified via `start` parameter. See [CoroutineStart] for details.
- * An optional [start] parameter can be set to [CoroutineStart.LAZY] to start coroutine _lazily_. In this case,,
- * the resulting [Deferred] is created in _new_ state. It can be explicitly started with [start][Job.start]
- * function and will be started implicitly on the first invocation of [join][Job.join] or [await][Deferred.await].
- *
- * @param context context of the coroutine. The default value is [DefaultDispatcher].
- * @param start coroutine start option. The default value is [CoroutineStart.DEFAULT].
- * @param parent explicitly specifies the parent job, overrides job from the [context] (if any).
- * @param onCompletion optional completion handler for the coroutine (see [Job.invokeOnCompletion]).
- * @param block the coroutine code.
- */
-public actual fun <T> async(
- context: CoroutineContext = DefaultDispatcher,
- start: CoroutineStart = CoroutineStart.DEFAULT,
- parent: Job? = null,
- onCompletion: CompletionHandler? = null,
- block: suspend CoroutineScope.() -> T
-): Deferred<T> {
- val newContext = newCoroutineContext(context, parent)
- val coroutine = if (start.isLazy)
- LazyDeferredCoroutine(newContext, block) else
- DeferredCoroutine<T>(newContext, active = true)
- if (onCompletion != null) coroutine.invokeOnCompletion(handler = onCompletion)
- coroutine.start(start, coroutine, block)
- return coroutine
-}
-
-/** @suppress **Deprecated**: Binary compatibility */
-@Deprecated(message = "Binary compatibility", level = DeprecationLevel.HIDDEN)
-public fun <T> async(
- context: CoroutineContext = DefaultDispatcher,
- start: CoroutineStart = CoroutineStart.DEFAULT,
- parent: Job? = null,
- block: suspend CoroutineScope.() -> T
-): Deferred<T> = async(context, start, parent, block = block)
-
-/** @suppress **Deprecated**: Binary compatibility */
-@Deprecated(message = "Binary compatibility", level = DeprecationLevel.HIDDEN)
-public fun <T> async(
- context: CoroutineContext = DefaultDispatcher,
- start: CoroutineStart = CoroutineStart.DEFAULT,
- block: suspend CoroutineScope.() -> T
-): Deferred<T> =
- async(context, start, block = block)
-
-/**
- * @suppress **Deprecated**: Use `start = CoroutineStart.XXX` parameter
- */
-@Deprecated(message = "Use `start = CoroutineStart.XXX` parameter",
- replaceWith = ReplaceWith("async(context, if (start) CoroutineStart.DEFAULT else CoroutineStart.LAZY, block)"))
-public fun <T> async(context: CoroutineContext, start: Boolean, block: suspend CoroutineScope.() -> T): Deferred<T> =
- async(context, if (start) CoroutineStart.DEFAULT else CoroutineStart.LAZY, block = block)
-
-/**
- * @suppress **Deprecated**: `defer` was renamed to `async`.
- */
-@Deprecated(message = "`defer` was renamed to `async`", level = DeprecationLevel.WARNING,
- replaceWith = ReplaceWith("async(context, block = block)"))
-public fun <T> defer(context: CoroutineContext, block: suspend CoroutineScope.() -> T): Deferred<T> =
- async(context, block = block)
-
-@Suppress("UNCHECKED_CAST")
-private open class DeferredCoroutine<T>(
- parentContext: CoroutineContext,
- active: Boolean
-) : AbstractCoroutine<T>(parentContext, active), Deferred<T>, SelectClause1<T> {
- override fun getCompleted(): T = getCompletedInternal() as T
- override suspend fun await(): T = awaitInternal() as T
- override val onAwait: SelectClause1<T> get() = this
- override fun <R> registerSelectClause1(select: SelectInstance<R>, block: suspend (T) -> R) =
- registerSelectClause1Internal(select, block)
-}
-
-private class LazyDeferredCoroutine<T>(
- parentContext: CoroutineContext,
- private val block: suspend CoroutineScope.() -> T
-) : DeferredCoroutine<T>(parentContext, active = false) {
- override fun onStart() {
- block.startCoroutineCancellable(this, this)
- }
-}
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Delay.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Delay.kt
deleted file mode 100644
index 8d1e01e..0000000
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Delay.kt
+++ /dev/null
@@ -1,124 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental
-
-import kotlinx.coroutines.experimental.selects.SelectBuilder
-import kotlinx.coroutines.experimental.selects.select
-import java.util.concurrent.Future
-import java.util.concurrent.TimeUnit
-import kotlin.coroutines.experimental.ContinuationInterceptor
-import kotlin.coroutines.experimental.CoroutineContext
-
-/**
- * This dispatcher _feature_ is implemented by [CoroutineDispatcher] implementations that natively support
- * scheduled execution of tasks.
- *
- * Implementation of this interface affects operation of
- * [delay][kotlinx.coroutines.experimental.delay] and [withTimeout] functions.
- */
-public actual interface Delay {
- /**
- * Delays coroutine for a given time without blocking a thread and resumes it after a specified time.
- * This suspending function is cancellable.
- * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
- * immediately resumes with [CancellationException].
- */
- suspend fun delay(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS) {
- require(time >= 0) { "Delay time $time cannot be negative" }
- if (time <= 0) return // don't delay
- return suspendCancellableCoroutine { scheduleResumeAfterDelay(time, unit, it) }
- }
-
- /**
- * Schedules resume of a specified [continuation] after a specified delay [time].
- *
- * Continuation **must be scheduled** to resume even if it is already cancelled, because a cancellation is just
- * an exception that the coroutine that used `delay` might wanted to catch and process. It might
- * need to close some resources in its `finally` blocks, for example.
- *
- * This implementation is supposed to use dispatcher's native ability for scheduled execution in its thread(s).
- * In order to avoid an extra delay of execution, the following code shall be used to resume this
- * [continuation] when the code is already executing in the appropriate thread:
- *
- * ```kotlin
- * with(continuation) { resumeUndispatched(Unit) }
- * ```
- */
- fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>)
-
- /**
- * Schedules invocation of a specified [block] after a specified delay [time].
- * The resulting [DisposableHandle] can be used to [dispose][DisposableHandle.dispose] of this invocation
- * request if it is not needed anymore.
- *
- * This implementation uses a built-in single-threaded scheduled executor service.
- */
- fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle =
- DefaultExecutor.invokeOnTimeout(time, unit, block)
-}
-
-/**
- * Delays coroutine for a given time without blocking a thread and resumes it after a specified time.
- * This suspending function is cancellable.
- * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
- * immediately resumes with [CancellationException].
- *
- * Note, that delay can be used in [select] invocation with [onTimeout][SelectBuilder.onTimeout] clause.
- *
- * This function delegates to [Delay.scheduleResumeAfterDelay] if the context [CoroutineDispatcher]
- * implements [Delay] interface, otherwise it resumes using a built-in single-threaded scheduled executor service.
- *
- * @param time time in milliseconds.
- */
-public actual suspend fun delay(time: Int) =
- delay(time.toLong(), TimeUnit.MILLISECONDS)
-
-/**
- * Delays coroutine for a given time without blocking a thread and resumes it after a specified time.
- * This suspending function is cancellable.
- * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
- * immediately resumes with [CancellationException].
- *
- * Note, that delay can be used in [select] invocation with [onTimeout][SelectBuilder.onTimeout] clause.
- *
- * This function delegates to [Delay.scheduleResumeAfterDelay] if the context [CoroutineDispatcher]
- * implements [Delay] interface, otherwise it resumes using a built-in single-threaded scheduled executor service.
- *
- * @param time time in the specified [unit].
- * @param unit time unit.
- */
-public suspend fun delay(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS) {
- require(time >= 0) { "Delay time $time cannot be negative" }
- if (time <= 0) return // don't delay
- return suspendCancellableCoroutine sc@ { cont: CancellableContinuation<Unit> ->
- cont.context.delay.scheduleResumeAfterDelay(time, unit, cont)
- }
-}
-
-/**
- * An implementation of [DisposableHandle] that cancels the specified future on dispose.
- * @suppress **This is unstable API and it is subject to change.**
- */
-public class DisposableFutureHandle(private val future: Future<*>) : DisposableHandle {
- override fun dispose() {
- future.cancel(false)
- }
- override fun toString(): String = "DisposableFutureHandle[$future]"
-}
-
-/** Returns [Delay] implementation of the given context */
-internal val CoroutineContext.delay: Delay get() = get(ContinuationInterceptor) as? Delay ?: DefaultExecutor
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/EventLoop.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/EventLoop.kt
index e752994..6d6685b 100644
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/EventLoop.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/EventLoop.kt
@@ -18,7 +18,7 @@
import kotlinx.atomicfu.*
import kotlinx.coroutines.experimental.internal.*
-import java.util.concurrent.*
+import kotlinx.coroutines.experimental.timeunit.TimeUnit
import java.util.concurrent.locks.*
import kotlin.coroutines.experimental.*
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Exceptions.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Exceptions.kt
index ed43104..7d7c532 100644
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Exceptions.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Exceptions.kt
@@ -26,9 +26,12 @@
cause: Throwable
) : RuntimeException(message, cause)
- /**
- * Thrown by cancellable suspending functions if the [Job] of the coroutine is cancelled while it is suspending.
- */
+/**
+ * Thrown by cancellable suspending functions if the [Job] of the coroutine is cancelled while it is suspending.
+ * It indicates _normal_ cancellation of a coroutine.
+ * **It is not printed to console/log by default uncaught exception handler**.
+ * (see [handleCoroutineException]).
+*/
public actual typealias CancellationException = java.util.concurrent.CancellationException
/**
@@ -53,31 +56,8 @@
(message!!.hashCode() * 31 + job.hashCode()) * 31 + (cause?.hashCode() ?: 0)
}
-/**
- * This exception is thrown by [withTimeout] to indicate timeout.
- */
-@Suppress("DEPRECATION")
-public actual class TimeoutCancellationException internal constructor(
- message: String,
- @JvmField internal val coroutine: Job?
-) : TimeoutException(message) {
- /**
- * Creates timeout exception with a given message.
- */
- public actual constructor(message: String) : this(message, null)
-}
-
-@Suppress("FunctionName")
-internal fun TimeoutCancellationException(
- time: Long,
- unit: TimeUnit,
- coroutine: Job
-) : TimeoutCancellationException = TimeoutCancellationException("Timed out waiting for $time $unit", coroutine)
-
-/**
- * @suppress **Deprecated**: Renamed to TimeoutCancellationException
- */
-@Deprecated("Renamed to TimeoutCancellationException", replaceWith = ReplaceWith("TimeoutCancellationException"))
-public open class TimeoutException(message: String) : CancellationException(message)
-
internal actual class DispatchException actual constructor(message: String, cause: Throwable) : RuntimeException(message, cause)
+
+@Suppress("NOTHING_TO_INLINE")
+internal actual inline fun Throwable.addSuppressedThrowable(other: Throwable) =
+ addSuppressed(other)
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Executors.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Executors.kt
index 545300f..f20753a 100644
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Executors.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Executors.kt
@@ -16,13 +16,10 @@
package kotlinx.coroutines.experimental
-import java.io.Closeable
-import java.util.concurrent.Executor
-import java.util.concurrent.ExecutorService
-import java.util.concurrent.RejectedExecutionException
-import java.util.concurrent.ScheduledExecutorService
-import java.util.concurrent.TimeUnit
-import kotlin.coroutines.experimental.CoroutineContext
+import kotlinx.coroutines.experimental.timeunit.TimeUnit
+import java.io.*
+import java.util.concurrent.*
+import kotlin.coroutines.experimental.*
/**
* [CoroutineDispatcher] that implements [Closeable]
@@ -108,3 +105,14 @@
with(continuation) { dispatcher.resumeUndispatched(Unit) }
}
}
+
+/**
+ * An implementation of [DisposableHandle] that cancels the specified future on dispose.
+ * @suppress **This is unstable API and it is subject to change.**
+ */
+public class DisposableFutureHandle(private val future: Future<*>) : DisposableHandle {
+ override fun dispose() {
+ future.cancel(false)
+ }
+ override fun toString(): String = "DisposableFutureHandle[$future]"
+}
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Future.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Future.kt
new file mode 100644
index 0000000..b10f0cd
--- /dev/null
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Future.kt
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+@file:JvmMultifileClass
+@file:JvmName("JobKt")
+
+package kotlinx.coroutines.experimental
+
+import java.util.concurrent.*
+
+/**
+ * Cancels a specified [future] when this job is complete.
+ *
+ * This is a shortcut for the following code with slightly more efficient implementation (one fewer object created).
+ * ```
+ * invokeOnCompletion { future.cancel(false) }
+ * ```
+ */
+public fun Job.cancelFutureOnCompletion(future: Future<*>): DisposableHandle =
+ invokeOnCompletion(handler = CancelFutureOnCompletion(this, future))
+
+private class CancelFutureOnCompletion(
+ job: Job,
+ private val future: Future<*>
+) : JobNode<Job>(job) {
+ override fun invoke(reason: Throwable?) {
+ // Don't interrupt when cancelling future on completion, because no one is going to reset this
+ // interruption flag and it will cause spurious failures elsewhere
+ future.cancel(false)
+ }
+ override fun toString() = "CancelFutureOnCompletion[$future]"
+}
+
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Job.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Job.kt
deleted file mode 100644
index 6c03c07..0000000
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Job.kt
+++ /dev/null
@@ -1,1521 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-// :todo: Remove after transition to Kotlin 1.2.30+
-@file:Suppress("ACTUAL_FUNCTION_WITH_DEFAULT_ARGUMENTS")
-
-package kotlinx.coroutines.experimental
-
-import kotlinx.atomicfu.*
-import kotlinx.coroutines.experimental.internal.*
-import kotlinx.coroutines.experimental.intrinsics.*
-import kotlinx.coroutines.experimental.selects.*
-import java.util.concurrent.*
-import kotlin.coroutines.experimental.*
-import kotlin.coroutines.experimental.intrinsics.*
-
-// --------------- core job interfaces ---------------
-
-/**
- * A background job. Conceptually, a job is a cancellable thing with a simple life-cycle that
- * culminates in its completion. Jobs can be arranged into parent-child hierarchies where cancellation
- * or completion of parent immediately cancels all its [children].
- *
- * The most basic instances of [Job] are created with [launch] coroutine builder or with a
- * `Job()` factory function. Other coroutine builders and primitives like
- * [Deferred] also implement [Job] interface.
- *
- * A job has the following states:
- *
- * | **State** | [isActive] | [isCompleted] | [isCancelled] |
- * | --------------------------------------- | ---------- | ------------- | ------------- |
- * | _New_ (optional initial state) | `false` | `false` | `false` |
- * | _Active_ (default initial state) | `true` | `false` | `false` |
- * | _Completing_ (optional transient state) | `true` | `false` | `false` |
- * | _Cancelling_ (optional transient state) | `false` | `false` | `true` |
- * | _Cancelled_ (final state) | `false` | `true` | `true` |
- * | _Completed_ (final state) | `false` | `true` | `false` |
- *
- * Usually, a job is created in _active_ state (it is created and started). However, coroutine builders
- * that provide an optional `start` parameter create a coroutine in _new_ state when this parameter is set to
- * [CoroutineStart.LAZY]. Such a job can be made _active_ by invoking [start] or [join].
- *
- * A job can be _cancelled_ at any time with [cancel] function that forces it to transition to
- * _cancelling_ state immediately. Job that is not backed by a coroutine (see `Job()` function) and does not have
- * [children] becomes _cancelled_ on [cancel] immediately.
- * Otherwise, job becomes _cancelled_ when it finishes executing its code and
- * when all its children [complete][isCompleted].
- *
- * ```
- * wait children
- * +-----+ start +--------+ complete +-------------+ finish +-----------+
- * | New | ---------------> | Active | -----------> | Completing | -------> | Completed |
- * +-----+ +--------+ +-------------+ +-----------+
- * | | |
- * | cancel | cancel | cancel
- * V V |
- * +-----------+ finish +------------+ |
- * | Cancelled | <--------- | Cancelling | <----------------+
- * |(completed)| +------------+
- * +-----------+
- * ```
- *
- * A job in the
- * [coroutineContext](https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.coroutines.experimental/coroutine-context.html)
- * represents the coroutine itself.
- * A job is active while the coroutine is working and job's cancellation aborts the coroutine when
- * the coroutine is suspended on a _cancellable_ suspension point by throwing [CancellationException].
- *
- * A job can have a _parent_ job. A job with a parent is cancelled when its parent is cancelled or completes exceptionally.
- * Parent job waits for all its children to complete in _completing_ or _cancelling_ state.
- * _Completing_ state is purely internal to the job. For an outside observer a _completing_ job is still active,
- * while internally it is waiting for its children.
- *
- * All functions on this interface and on all interfaces derived from it are **thread-safe** and can
- * be safely invoked from concurrent coroutines without external synchronization.
- */
-public actual interface Job : CoroutineContext.Element {
- /**
- * Key for [Job] instance in the coroutine context.
- */
- public actual companion object Key : CoroutineContext.Key<Job> {
- /**
- * Creates a new job object in _active_ state.
- * It is optionally a child of a [parent] job.
- * @suppress **Deprecated**
- */
- @Deprecated("Replaced with top-level function", level = DeprecationLevel.HIDDEN)
- public operator fun invoke(parent: Job? = null): Job = Job(parent)
-
- init {
- /*
- * Here we make sure that CoroutineExceptionHandler is always initialized in advance, so
- * that if a coroutine fails due to StackOverflowError we don't fail to report this error
- * trying to initialize CoroutineExceptionHandler
- */
- CoroutineExceptionHandler
- }
- }
-
- // ------------ state query ------------
-
- /**
- * Returns `true` when this job is active -- it was already started and has not completed or cancelled yet.
- * The job that is waiting for its [children] to complete is still considered to be active if it
- * was not cancelled.
- */
- public actual val isActive: Boolean
-
- /**
- * Returns `true` when this job has completed for any reason. A job that was cancelled and has
- * finished its execution is also considered complete. Job becomes complete only after
- * all its [children] complete.
- */
- public actual val isCompleted: Boolean
-
- /**
- * Returns `true` if this job was [cancelled][cancel]. In the general case, it does not imply that the
- * job has already [completed][isCompleted] (it may still be cancelling whatever it was doing).
- */
- public actual val isCancelled: Boolean
-
- /**
- * Returns [CancellationException] that signals the completion of this job. This function is
- * used by [cancellable][suspendCancellableCoroutine] suspending functions. They throw exception
- * returned by this function when they suspend in the context of this job and this job becomes _complete_.
- *
- * This function returns the original [cancel] cause of this job if that `cause` was an instance of
- * [CancellationException]. Otherwise (if this job was cancelled with a cause of a different type, or
- * was cancelled without a cause, or had completed normally), an instance of [JobCancellationException] is
- * returned. The [JobCancellationException.cause] of the resulting [JobCancellationException] references
- * the original cancellation cause that was passed to [cancel] function.
- *
- * This function throws [IllegalStateException] when invoked on a job that has not
- * [completed][isCompleted] nor [cancelled][isCancelled] yet.
- */
- public actual fun getCancellationException(): CancellationException
-
- /**
- * @suppress **Deprecated**: Renamed to [getCancellationException]
- */
- @Deprecated("Renamed to getCancellationException", replaceWith = ReplaceWith("getCancellationException()"))
- public fun getCompletionException(): Throwable =
- getCancellationException()
-
- // ------------ state update ------------
-
- /**
- * Starts coroutine related to this job (if any) if it was not started yet.
- * The result `true` if this invocation actually started coroutine or `false`
- * if it was already started or completed.
- */
- public actual fun start(): Boolean
-
- /**
- * Cancels this job with an optional cancellation [cause]. The result is `true` if this job was
- * cancelled as a result of this invocation and `false` otherwise
- * (if it was already _completed_ or if it is [NonCancellable]).
- * Repeated invocations of this function have no effect and always produce `false`.
- *
- * When cancellation has a clear reason in the code, an instance of [CancellationException] should be created
- * at the corresponding original cancellation site and passed into this method to aid in debugging by providing
- * both the context of cancellation and text description of the reason.
- */
- public actual fun cancel(cause: Throwable? = null): Boolean
-
- // ------------ parent-child ------------
-
- /**
- * Returns a sequence of this job's children.
- *
- * A job becomes a child of this job when it is constructed with this job in its
- * [CoroutineContext] or using an explicit `parent` parameter.
- *
- * A parent-child relation has the following effect:
- *
- * * Cancellation of parent with [cancel] or its exceptional completion (failure)
- * immediately cancels all its children.
- * * Parent cannot complete until all its children are complete. Parent waits for all its children to
- * complete in _completing_ or _cancelling_ state.
- * * Uncaught exception in a child, by default, cancels parent. In particular, this applies to
- * children created with [launch] coroutine builder. Note, that [async] and other future-like
- * coroutine builders do not have uncaught exceptions by definition, since all their exceptions are
- * caught and are encapsulated in their result.
- */
- public actual val children: Sequence<Job>
-
- /**
- * Attaches child job so that this job becomes its parent and
- * returns a handle that should be used to detach it.
- *
- * A parent-child relation has the following effect:
- * * Cancellation of parent with [cancel] or its exceptional completion (failure)
- * immediately cancels all its children.
- * * Parent cannot complete until all its children are complete. Parent waits for all its children to
- * complete in _completing_ or _cancelling_ state.
- *
- * **A child must store the resulting [DisposableHandle] and [dispose][DisposableHandle.dispose] the attachment
- * to its parent on its own completion.**
- *
- * Coroutine builders and job factory functions that accept `parent` [CoroutineContext] parameter
- * lookup a [Job] instance in the parent context and use this function to attach themselves as a child.
- * They also store a reference to the resulting [DisposableHandle] and dispose a handle when they complete.
- *
- * @suppress This is an internal API. This method is too error prone for public API.
- */
- @Deprecated(message = "Start child coroutine with 'parent' parameter", level = DeprecationLevel.WARNING)
- public actual fun attachChild(child: Job): DisposableHandle
-
- /**
- * Cancels all children jobs of this coroutine with the given [cause]. Unlike [cancel],
- * the state of this job itself is not affected.
- * @suppress **Deprecated**: Binary compatibility, it is an extension now
- */
- @Deprecated(message = "Binary compatibility, it is an extension now", level = DeprecationLevel.HIDDEN)
- public fun cancelChildren(cause: Throwable? = null)
-
- // ------------ state waiting ------------
-
- /**
- * Suspends coroutine until this job is complete. This invocation resumes normally (without exception)
- * when the job is complete for any reason and the [Job] of the invoking coroutine is still [active][isActive].
- * This function also [starts][Job.start] the corresponding coroutine if the [Job] was still in _new_ state.
- *
- * Note, that the job becomes complete only when all its children are complete.
- *
- * This suspending function is cancellable and **always** checks for the cancellation of invoking coroutine's Job.
- * If the [Job] of the invoking coroutine is cancelled or completed when this
- * suspending function is invoked or while it is suspended, this function
- * throws [CancellationException].
- *
- * In particular, it means that a parent coroutine invoking `join` on a child coroutine that was started using
- * `launch(coroutineContext) { ... }` builder throws [CancellationException] if the child
- * had crashed, unless a non-standard [CoroutineExceptionHandler] if installed in the context.
- *
- * This function can be used in [select] invocation with [onJoin] clause.
- * Use [isCompleted] to check for completion of this job without waiting.
- *
- * There is [cancelAndJoin] function that combines an invocation of [cancel] and `join`.
- */
- public actual suspend fun join()
-
- /**
- * Clause for [select] expression of [join] suspending function that selects when the job is complete.
- * This clause never fails, even if the job completes exceptionally.
- */
- public val onJoin: SelectClause0
-
- // ------------ low-level state-notification ------------
-
- /**
- * @suppress **Deprecated**: For binary compatibility
- */
- @Deprecated(message = "For binary compatibility", level = DeprecationLevel.HIDDEN)
- public fun invokeOnCompletion(handler: CompletionHandler, onCancelling: Boolean): DisposableHandle
-
- /**
- * @suppress **Deprecated**: Use with named `onCancelling` and `handler` parameters.
- */
- @Deprecated(message = "Use with named `onCancelling` and `handler` parameters", level = DeprecationLevel.WARNING,
- replaceWith = ReplaceWith("this.invokeOnCompletion(onCancelling = onCancelling_, handler = handler)"))
- public fun invokeOnCompletion(onCancelling_: Boolean = false, handler: CompletionHandler): DisposableHandle
-
- /**
- * Registers handler that is **synchronously** invoked once on completion of this job.
- * When job is already complete, then the handler is immediately invoked
- * with a job's exception or cancellation cause or `null`. Otherwise, handler will be invoked once when this
- * job is complete.
- *
- * The resulting [DisposableHandle] can be used to [dispose][DisposableHandle.dispose] the
- * registration of this handler and release its memory if its invocation is no longer needed.
- * There is no need to dispose the handler after completion of this job. The references to
- * all the handlers are released when this job completes.
- *
- * Installed [handler] should not throw any exceptions. If it does, they will get caught,
- * wrapped into [CompletionHandlerException], and rethrown, potentially causing crash of unrelated code.
- *
- * **Note**: Implementations of `CompletionHandler` must be fast and _lock-free_.
- */
- public fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle
-
- /**
- * Registers handler that is **synchronously** invoked once on cancellation or completion of this job.
- * When job is already cancelling or complete, then the handler is immediately invoked
- * with a job's cancellation cause or `null` unless [invokeImmediately] is set to false.
- * Otherwise, handler will be invoked once when this job is cancelled or complete.
- *
- * Invocation of this handler on a transition to a transient _cancelling_ state
- * is controlled by [onCancelling] boolean parameter.
- * The handler is invoked on invocation of [cancel] when
- * job becomes _cancelling_ if [onCancelling] parameter is set to `true`. However,
- * when this [Job] is not backed by a coroutine, like [CompletableDeferred] or [CancellableContinuation]
- * (both of which do not posses a _cancelling_ state), then the value of [onCancelling] parameter is ignored.
- *
- * The resulting [DisposableHandle] can be used to [dispose][DisposableHandle.dispose] the
- * registration of this handler and release its memory if its invocation is no longer needed.
- * There is no need to dispose the handler after completion of this job. The references to
- * all the handlers are released when this job completes.
- *
- * Installed [handler] should not throw any exceptions. If it does, they will get caught,
- * wrapped into [CompletionHandlerException], and rethrown, potentially causing crash of unrelated code.
- *
- * **Note**: This function is a part of internal machinery that supports parent-child hierarchies
- * and allows for implementation of suspending functions that wait on the Job's state.
- * This function should not be used in general application code.
- * Implementations of `CompletionHandler` must be fast and _lock-free_.
- *
- * @param onCancelling when `true`, then the [handler] is invoked as soon as this job transitions to _cancelling_ state;
- * when `false` then the [handler] is invoked only when it transitions to _completed_ state.
- * @param invokeImmediately when `true` and this job is already in the desired state (depending on [onCancelling]),
- * then the [handler] is immediately and synchronously invoked and [NonDisposableHandle] is returned;
- * when `false` then [NonDisposableHandle] is returned, but the [handler] is not invoked.
- * @param handler the handler.
- */
- public actual fun invokeOnCompletion(
- onCancelling: Boolean = false,
- invokeImmediately: Boolean = true,
- handler: CompletionHandler): DisposableHandle
-
- // ------------ unstable internal API ------------
-
- /**
- * @suppress **Error**: Operator '+' on two Job objects is meaningless.
- * Job is a coroutine context element and `+` is a set-sum operator for coroutine contexts.
- * The job to the right of `+` just replaces the job the left of `+`.
- */
- @Suppress("DeprecatedCallableAddReplaceWith")
- @Deprecated(message = "Operator '+' on two Job objects is meaningless. " +
- "Job is a coroutine context element and `+` is a set-sum operator for coroutine contexts. " +
- "The job to the right of `+` just replaces the job the left of `+`.",
- level = DeprecationLevel.ERROR)
- public operator fun plus(other: Job) = other
-
- /**
- * Registration object for [invokeOnCompletion]. It can be used to [unregister] if needed.
- * There is no need to unregister after completion.
- * @suppress **Deprecated**: Replace with `DisposableHandle`
- */
- @Deprecated(message = "Replace with `DisposableHandle`",
- replaceWith = ReplaceWith("DisposableHandle"))
- public interface Registration {
- /**
- * Unregisters completion handler.
- * @suppress **Deprecated**: Replace with `dispose`
- */
- @Deprecated(message = "Replace with `dispose`",
- replaceWith = ReplaceWith("dispose()"))
- public fun unregister()
- }
-}
-
-/**
- * Creates a new job object in an _active_ state.
- * It is optionally a child of a [parent] job.
- */
-@Suppress("FunctionName")
-public actual fun Job(parent: Job? = null): Job = JobImpl(parent)
-
-/**
- * A handle to an allocated object that can be disposed to make it eligible for garbage collection.
- */
-@Suppress("DEPRECATION") // todo: remove when Job.Registration is removed
-public actual interface DisposableHandle : Job.Registration {
- /**
- * Disposes the corresponding object, making it eligible for garbage collection.
- * Repeated invocation of this function has no effect.
- */
- public actual fun dispose()
-
- /**
- * Unregisters completion handler.
- * @suppress **Deprecated**: Replace with `dispose`
- */
- @Deprecated(message = "Replace with `dispose`",
- replaceWith = ReplaceWith("dispose()"))
- public override fun unregister() = dispose()
-}
-
-// -------------------- Job extensions --------------------
-
-/**
- * Unregisters a specified [registration] when this job is complete.
- *
- * This is a shortcut for the following code with slightly more efficient implementation (one fewer object created).
- * ```
- * invokeOnCompletion { registration.unregister() }
- * ```
- * @suppress: **Deprecated**: Renamed to `disposeOnCompletion`.
- */
-@Deprecated(message = "Renamed to `disposeOnCompletion`",
- replaceWith = ReplaceWith("disposeOnCompletion(registration)"))
-public fun Job.unregisterOnCompletion(registration: DisposableHandle): DisposableHandle =
- invokeOnCompletion(handler = DisposeOnCompletion(this, registration))
-
-/**
- * Disposes a specified [handle] when this job is complete.
- *
- * This is a shortcut for the following code with slightly more efficient implementation (one fewer object created).
- * ```
- * invokeOnCompletion { handle.dispose() }
- * ```
- */
-public actual fun Job.disposeOnCompletion(handle: DisposableHandle): DisposableHandle =
- invokeOnCompletion(handler = DisposeOnCompletion(this, handle))
-
-/**
- * Cancels a specified [future] when this job is complete.
- *
- * This is a shortcut for the following code with slightly more efficient implementation (one fewer object created).
- * ```
- * invokeOnCompletion { future.cancel(false) }
- * ```
- */
-public fun Job.cancelFutureOnCompletion(future: Future<*>): DisposableHandle =
- invokeOnCompletion(handler = CancelFutureOnCompletion(this, future))
-
-/**
- * Cancels the job and suspends invoking coroutine until the cancelled job is complete.
- *
- * This suspending function is cancellable and **always** checks for the cancellation of invoking coroutine's Job.
- * If the [Job] of the invoking coroutine is cancelled or completed when this
- * suspending function is invoked or while it is suspended, this function
- * throws [CancellationException].
- *
- * In particular, it means that a parent coroutine invoking `cancelAndJoin` on a child coroutine that was started using
- * `launch(coroutineContext) { ... }` builder throws [CancellationException] if the child
- * had crashed, unless a non-standard [CoroutineExceptionHandler] if installed in the context.
- *
- * This is a shortcut for the invocation of [cancel][Job.cancel] followed by [join][Job.join].
- */
-public actual suspend fun Job.cancelAndJoin() {
- cancel()
- return join()
-}
-
-/**
- * Cancels all [children][Job.children] jobs of this coroutine with the given [cause] using [Job.cancel]
- * for all of them. Unlike [Job.cancel] on this job as a whole, the state of this job itself is not affected.
- */
-@Suppress("EXTENSION_SHADOWED_BY_MEMBER") // See KT-21598
-public actual fun Job.cancelChildren(cause: Throwable? = null) {
- children.forEach { it.cancel(cause) }
-}
-
-/**
- * Suspends coroutine until all [children][Job.children] of this job are complete using
- * [Job.join] for all of them. Unlike [Job.join] on this job as a whole, it does not wait until
- * this job is complete.
- */
-public actual suspend fun Job.joinChildren() {
- children.forEach { it.join() }
-}
-
-// -------------------- CoroutineContext extensions --------------------
-
-/**
- * Returns `true` when the [Job] of the coroutine in this context is still active
- * (has not completed and was not cancelled yet).
- *
- * Check this property in long-running computation loops to support cancellation
- * when [CoroutineScope.isActive] is not available:
- *
- * ```
- * while (coroutineContext.isActive) {
- * // do some computation
- * }
- * ```
- *
- * The `coroutineContext.isActive` expression is a shortcut for `coroutineContext[Job]?.isActive == true`.
- * See [Job.isActive].
- */
-public actual val CoroutineContext.isActive: Boolean
- get() = this[Job]?.isActive == true
-
-/**
- * Cancels [Job] of this context with an optional cancellation [cause]. The result is `true` if the job was
- * cancelled as a result of this invocation and `false` if there is no job in the context or if it was already
- * cancelled or completed. See [Job.cancel] for details.
- */
-public actual fun CoroutineContext.cancel(cause: Throwable? = null): Boolean =
- this[Job]?.cancel(cause) ?: false
-
-/**
- * Cancels all children of the [Job] in this context with an optional cancellation [cause].
- * It does not do anything if there is no job in the context or it has no children.
- * See [Job.cancelChildren] for details.
- */
-public actual fun CoroutineContext.cancelChildren(cause: Throwable? = null) {
- this[Job]?.cancelChildren(cause)
-}
-
-/**
- * @suppress **Deprecated**: `join` is now a member function of `Job`.
- */
-@Suppress("EXTENSION_SHADOWED_BY_MEMBER", "DeprecatedCallableAddReplaceWith")
-@Deprecated(message = "`join` is now a member function of `Job`")
-public suspend fun Job.join() = this.join()
-
-/**
- * No-op implementation of [Job.Registration].
- * @suppress: **Deprecated**: Replace with [NonDisposableHandle]
- */
-@Deprecated(message = "Replace with `NonDisposableHandle`",
- replaceWith = ReplaceWith("NonDisposableHandle"))
-@Suppress("unused")
-typealias EmptyRegistration = NonDisposableHandle
-
-/**
- * No-op implementation of [DisposableHandle].
- */
-public actual object NonDisposableHandle : DisposableHandle {
- /** Does not do anything. */
- actual override fun dispose() {}
-
- /** Returns "NonDisposableHandle" string. */
- override fun toString(): String = "NonDisposableHandle"
-}
-
-// --------------- helper classes to simplify job implementation
-
-/**
- * A concrete implementation of [Job]. It is optionally a child to a parent job.
- * This job is cancelled when the parent is complete, but not vise-versa.
- *
- * This is an open class designed for extension by more specific classes that might augment the
- * state and mare store addition state information for completed jobs, like their result values.
- *
- * @param active when `true` the job is created in _active_ state, when `false` in _new_ state. See [Job] for details.
- * @suppress **This is unstable API and it is subject to change.**
- */
-internal actual open class JobSupport actual constructor(active: Boolean) : Job, SelectClause0 {
- actual final override val key: CoroutineContext.Key<*> get() = Job
-
- /*
- === Internal states ===
-
- name state class public state description
- ------ ------------ ------------ -----------
- EMPTY_N EmptyNew : New no listeners
- EMPTY_A EmptyActive : Active no listeners
- SINGLE JobNode : Active a single listener
- SINGLE+ JobNode : Active a single listener + NodeList added as its next
- LIST_N NodeList : New a list of listeners (promoted once, does not got back to EmptyNew)
- LIST_A NodeList : Active a list of listeners (promoted once, does not got back to JobNode/EmptyActive)
- COMPLETING Finishing : Completing has a list of listeners (promoted once from LIST_*)
- CANCELLING Finishing : Cancelling has a list of listeners (promoted once from LIST_*)
- FINAL_C Cancelled : Cancelled cancelled (final state)
- FINAL_F Failed : Completed failed for other reason (final state)
- FINAL_R <any> : Completed produced some result
-
- === Transitions ===
-
- New states Active states Inactive states
-
- +---------+ +---------+ }
- | EMPTY_N | --+-> | EMPTY_A | ----+ } Empty states
- +---------+ | +---------+ | }
- | | | ^ | +----------+
- | | | | +--> | FINAL_* |
- | | V | | +----------+
- | | +---------+ | }
- | | | SINGLE | ----+ } JobNode states
- | | +---------+ | }
- | | | | }
- | | V | }
- | | +---------+ | }
- | +-- | SINGLE+ | ----+ }
- | +---------+ | }
- | | |
- V V |
- +---------+ +---------+ | }
- | LIST_N | ----> | LIST_A | ----+ } NodeList states
- +---------+ +---------+ | }
- | | | | |
- | | +--------+ | |
- | | | V |
- | | | +------------+ | +------------+ }
- | +-------> | COMPLETING | --+-- | CANCELLING | } Finishing states
- | | +------------+ +------------+ }
- | | | ^
- | | | |
- +--------+---------+--------------------+
-
-
- This state machine and its transition matrix are optimized for the common case when job is created in active
- state (EMPTY_A) and at most one completion listener is added to it during its life-time.
-
- Note, that the actual `_state` variable can also be a reference to atomic operation descriptor `OpDescriptor`
- */
-
- // Note: use shared objects while we have no listeners
- private val _state = atomic<Any?>(if (active) EmptyActive else EmptyNew)
-
- @Volatile
- private var parentHandle: DisposableHandle? = null
-
- // ------------ initialization ------------
-
- /**
- * Initializes parent job.
- * It shall be invoked at most once after construction after all other initialization.
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal actual fun initParentJobInternal(parent: Job?) {
- check(parentHandle == null)
- if (parent == null) {
- parentHandle = NonDisposableHandle
- return
- }
- parent.start() // make sure the parent is started
- @Suppress("DEPRECATION")
- val handle = parent.attachChild(this)
- parentHandle = handle
- // now check our state _after_ registering (see updateState order of actions)
- if (isCompleted) {
- handle.dispose()
- parentHandle = NonDisposableHandle // release it just in case, to aid GC
- }
- }
-
- // ------------ state query ------------
-
- /**
- * Returns current state of this job.
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal val state: Any? get() {
- _state.loop { state -> // helper loop on state (complete in-progress atomic operations)
- if (state !is OpDescriptor) return state
- state.perform(this)
- }
- }
-
- /**
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal inline fun loopOnState(block: (Any?) -> Unit): Nothing {
- while (true) {
- block(state)
- }
- }
-
- public actual final override val isActive: Boolean get() {
- val state = this.state
- return state is Incomplete && state.isActive
- }
-
- public actual final override val isCompleted: Boolean get() = state !is Incomplete
-
- public actual final override val isCancelled: Boolean get() {
- val state = this.state
- return state is Cancelled || (state is Finishing && state.cancelled != null)
- }
-
- // ------------ state update ------------
-
- /**
- * Updates current [state] of this job. Returns `false` if current state is not equal to expected.
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal fun updateState(expect: Incomplete, proposedUpdate: Any?, mode: Int): Boolean {
- val update = coerceProposedUpdate(expect, proposedUpdate)
- if (!tryUpdateState(expect, update)) return false
- completeUpdateState(expect, update, mode)
- return true
- }
-
- // when Job is in Cancelling state, it can only be promoted to Cancelled state,
- // so if the proposed Update is not an appropriate Cancelled (preserving the cancellation cause),
- // then the corresponding Cancelled state is constructed.
- private fun coerceProposedUpdate(expect: Incomplete, proposedUpdate: Any?): Any? =
- if (expect is Finishing && expect.cancelled != null && !isCorrespondinglyCancelled(expect.cancelled, proposedUpdate))
- createCancelled(expect.cancelled, proposedUpdate) else proposedUpdate
-
- private fun isCorrespondinglyCancelled(cancelled: Cancelled, proposedUpdate: Any?): Boolean {
- if (proposedUpdate !is Cancelled) return false
- // NOTE: equality comparison of causes is performed here by design, see equals of JobCancellationException
- return proposedUpdate.cause == cancelled.cause ||
- proposedUpdate.cause is JobCancellationException && cancelled.cause == null
- }
-
- private fun createCancelled(cancelled: Cancelled, proposedUpdate: Any?): Cancelled {
- if (proposedUpdate !is CompletedExceptionally) return cancelled // not exception -- just use original cancelled
- val exception = proposedUpdate.exception
- if (cancelled.exception == exception) return cancelled // that is the cancelled we need already!
- cancelled.cause?.let { exception.addSuppressed(it) }
- return Cancelled(this, exception)
- }
-
- /**
- * Tries to initiate update of the current [state] of this job.
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal fun tryUpdateState(expect: Incomplete, update: Any?): Boolean {
- require(update !is Incomplete) // only incomplete -> completed transition is allowed
- if (!_state.compareAndSet(expect, update)) return false
- // Unregister from parent job
- parentHandle?.let {
- it.dispose() // volatile read parentHandle _after_ state was updated
- parentHandle = NonDisposableHandle // release it just in case, to aid GC
- }
- return true // continues in completeUpdateState
- }
-
- /**
- * Completes update of the current [state] of this job.
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal fun completeUpdateState(expect: Incomplete, update: Any?, mode: Int) {
- val exceptionally = update as? CompletedExceptionally
- // Do overridable processing before completion handlers
- if (!expect.isCancelling) onCancellationInternal(exceptionally) // only notify when was not cancelling before
- onCompletionInternal(update, mode)
- // Invoke completion handlers
- val cause = exceptionally?.cause
- if (expect is JobNode<*>) { // SINGLE/SINGLE+ state -- one completion handler (common case)
- try {
- expect.invoke(cause)
- } catch (ex: Throwable) {
- handleException(CompletionHandlerException("Exception in completion handler $expect for $this", ex))
- }
- } else {
- expect.list?.notifyCompletion(cause)
- }
- }
-
- private inline fun <reified T: JobNode<*>> notifyHandlers(list: NodeList, cause: Throwable?) {
- var exception: Throwable? = null
- list.forEach<T> { node ->
- try {
- node.invoke(cause)
- } catch (ex: Throwable) {
- exception?.apply { addSuppressed(ex) } ?: run {
- exception = CompletionHandlerException("Exception in completion handler $node for $this", ex)
- }
- }
- }
- exception?.let { handleException(it) }
- }
-
- private fun NodeList.notifyCompletion(cause: Throwable?) =
- notifyHandlers<JobNode<*>>(this, cause)
-
- private fun notifyCancellation(list: NodeList, cause: Throwable?) =
- notifyHandlers<JobCancellationNode<*>>(list, cause)
-
- public actual final override fun start(): Boolean {
- loopOnState { state ->
- when (startInternal(state)) {
- FALSE -> return false
- TRUE -> return true
- }
- }
- }
-
- // returns: RETRY/FALSE/TRUE:
- // FALSE when not new,
- // TRUE when started
- // RETRY when need to retry
- private fun startInternal(state: Any?): Int {
- when (state) {
- is Empty -> { // EMPTY_X state -- no completion handlers
- if (state.isActive) return FALSE // already active
- if (!_state.compareAndSet(state, EmptyActive)) return RETRY
- onStartInternal()
- return TRUE
- }
- is NodeList -> { // LIST -- a list of completion handlers (either new or active)
- return state.tryMakeActive().also { result ->
- if (result == TRUE) onStartInternal()
- }
- }
- else -> return FALSE // not a new state
- }
- }
-
- /**
- * Override to provide the actual [start] action.
- * This function is invoked exactly once when non-active coroutine is [started][start].
- */
- internal actual open fun onStartInternal() {}
-
- public actual final override fun getCancellationException(): CancellationException {
- val state = this.state
- return when {
- state is Finishing && state.cancelled != null ->
- state.cancelled.exception.toCancellationException("Job is being cancelled")
- state is Incomplete ->
- error("Job was not completed or cancelled yet: $this")
- state is CompletedExceptionally ->
- state.exception.toCancellationException("Job has failed")
- else -> JobCancellationException("Job has completed normally", null, this)
- }
- }
-
- private fun Throwable.toCancellationException(message: String): CancellationException =
- this as? CancellationException ?: JobCancellationException(message, this, this@JobSupport)
-
- /**
- * Returns the cause that signals the completion of this job -- it returns the original
- * [cancel] cause or **`null` if this job had completed
- * normally or was cancelled without a cause**. This function throws
- * [IllegalStateException] when invoked for an job that has not [completed][isCompleted] nor
- * [isCancelled] yet.
- */
- protected fun getCompletionCause(): Throwable? {
- val state = this.state
- return when {
- state is Finishing && state.cancelled != null -> state.cancelled.cause
- state is Incomplete -> error("Job was not completed or cancelled yet")
- state is CompletedExceptionally -> state.cause
- else -> null
- }
- }
-
- @Suppress("OverridingDeprecatedMember")
- public final override fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle =
- invokeOnCompletion(onCancelling = false, invokeImmediately = true, handler = handler)
-
- @Suppress("OverridingDeprecatedMember")
- public final override fun invokeOnCompletion(handler: CompletionHandler, onCancelling: Boolean): DisposableHandle =
- invokeOnCompletion(onCancelling = onCancelling, invokeImmediately = true, handler = handler)
-
- @Suppress("OverridingDeprecatedMember")
- public final override fun invokeOnCompletion(onCancelling_: Boolean, handler: CompletionHandler): DisposableHandle =
- invokeOnCompletion(onCancelling = onCancelling_, invokeImmediately = true, handler = handler)
-
- // todo: non-final as a workaround for KT-21968, should be final in the future
- public actual override fun invokeOnCompletion(
- onCancelling: Boolean,
- invokeImmediately: Boolean,
- handler: CompletionHandler
- ): DisposableHandle {
- var nodeCache: JobNode<*>? = null
- loopOnState { state ->
- when (state) {
- is Empty -> { // EMPTY_X state -- no completion handlers
- if (state.isActive) {
- // try move to SINGLE state
- val node = nodeCache ?: makeNode(handler, onCancelling).also { nodeCache = it }
- if (_state.compareAndSet(state, node)) return node
- } else
- promoteEmptyToNodeList(state) // that way we can add listener for non-active coroutine
- }
- is Incomplete -> {
- val list = state.list
- if (list == null) { // SINGLE/SINGLE+
- promoteSingleToNodeList(state as JobNode<*>)
- } else {
- if (state is Finishing && state.cancelled != null && onCancelling) {
- check(onCancelMode != ON_CANCEL_MAKE_CANCELLED) // cannot be in this state unless were support cancelling state
- // installing cancellation handler on job that is being cancelled
- if (invokeImmediately) handler(state.cancelled.cause)
- return NonDisposableHandle
- }
- val node = nodeCache ?: makeNode(handler, onCancelling).also { nodeCache = it }
- if (addLastAtomic(state, list, node)) return node
- }
- }
- else -> { // is complete
- if (invokeImmediately) handler((state as? CompletedExceptionally)?.cause)
- return NonDisposableHandle
- }
- }
- }
- }
-
- private fun makeNode(handler: CompletionHandler, onCancelling: Boolean): JobNode<*> {
- val hasCancellingState = onCancelMode != ON_CANCEL_MAKE_CANCELLED
- return if (onCancelling && hasCancellingState)
- (handler as? JobCancellationNode<*>)?.also { require(it.job === this) }
- ?: InvokeOnCancellation(this, handler)
- else
- (handler as? JobNode<*>)?.also { require(it.job === this && (!hasCancellingState || it !is JobCancellationNode)) }
- ?: InvokeOnCompletion(this, handler)
- }
-
- private fun addLastAtomic(expect: Any, list: NodeList, node: JobNode<*>) =
- list.addLastIf(node) { this.state === expect }
-
- private fun promoteEmptyToNodeList(state: Empty) {
- // try to promote it to list in new state
- _state.compareAndSet(state, NodeList(state.isActive))
- }
-
- private fun promoteSingleToNodeList(state: JobNode<*>) {
- // try to promote it to list (SINGLE+ state)
- state.addOneIfEmpty(NodeList(active = true))
- // it must be in SINGLE+ state or state has changed (node could have need removed from state)
- val list = state.next // either NodeList or somebody else won the race, updated state
- // just attempt converting it to list if state is still the same, then we'll continue lock-free loop
- _state.compareAndSet(state, list)
- }
-
- public actual final override suspend fun join() {
- if (!joinInternal()) { // fast-path no wait
- return suspendCoroutineOrReturn { cont ->
- cont.context.checkCompletion()
- Unit // do not suspend
- }
- }
- return joinSuspend() // slow-path wait
- }
-
- private fun joinInternal(): Boolean {
- loopOnState { state ->
- if (state !is Incomplete) return false // not active anymore (complete) -- no need to wait
- if (startInternal(state) >= 0) return true // wait unless need to retry
- }
- }
-
- private suspend fun joinSuspend() = suspendCancellableCoroutine<Unit> { cont ->
- cont.disposeOnCompletion(invokeOnCompletion(handler = ResumeOnCompletion(this, cont)))
- }
-
- public final override val onJoin: SelectClause0
- get() = this
-
- // registerSelectJoin
- public final override fun <R> registerSelectClause0(select: SelectInstance<R>, block: suspend () -> R) {
- // fast-path -- check state and select/return if needed
- loopOnState { state ->
- if (select.isSelected) return
- if (state !is Incomplete) {
- // already complete -- select result
- if (select.trySelect(null)) {
- select.completion.context.checkCompletion() // always check for our completion
- block.startCoroutineUndispatched(select.completion)
- }
- return
- }
- if (startInternal(state) == 0) {
- // slow-path -- register waiter for completion
- select.disposeOnSelect(invokeOnCompletion(handler = SelectJoinOnCompletion(this, select, block)))
- return
- }
- }
- }
-
- /**
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal fun removeNode(node: JobNode<*>) {
- // remove logic depends on the state of the job
- loopOnState { state ->
- when (state) {
- is JobNode<*> -> { // SINGE/SINGLE+ state -- one completion handler
- if (state !== node) return // a different job node --> we were already removed
- // try remove and revert back to empty state
- if (_state.compareAndSet(state, EmptyActive)) return
- }
- is Incomplete -> { // may have a list of completion handlers
- // remove node from the list if there is a list
- if (state.list != null) node.remove()
- return
- }
- else -> return // it is complete and does not have any completion handlers
- }
- }
- }
-
- /**
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal open val onCancelMode: Int get() = ON_CANCEL_MAKE_CANCELLING
-
- public actual override fun cancel(cause: Throwable?): Boolean = when (onCancelMode) {
- ON_CANCEL_MAKE_CANCELLED -> makeCancelled(cause)
- ON_CANCEL_MAKE_CANCELLING -> makeCancelling(cause)
- ON_CANCEL_MAKE_COMPLETING -> makeCompletingOnCancel(cause)
- else -> error("Invalid onCancelMode $onCancelMode")
- }
-
- // we will be dispatching coroutine to process its cancellation exception, so there is no need for
- // an extra check for Job status in MODE_CANCELLABLE
- private fun updateStateCancelled(state: Incomplete, cause: Throwable?) =
- updateState(state, Cancelled(this, cause), mode = MODE_ATOMIC_DEFAULT)
-
- // transitions to Cancelled state
- private fun makeCancelled(cause: Throwable?): Boolean {
- loopOnState { state ->
- if (state !is Incomplete) return false // quit if already complete
- if (updateStateCancelled(state, cause)) return true
- }
- }
-
- // transitions to Cancelling state
- private fun makeCancelling(cause: Throwable?): Boolean {
- loopOnState { state ->
- when (state) {
- is Empty -> { // EMPTY_X state -- no completion handlers
- if (state.isActive) {
- promoteEmptyToNodeList(state) // this way can wrap it into Cancelling on next pass
- } else {
- // cancelling a non-started coroutine makes it immediately cancelled
- // (and we have no listeners to notify which makes it very simple)
- if (updateStateCancelled(state, cause)) return true
- }
- }
- is JobNode<*> -> { // SINGLE/SINGLE+ state -- one completion handler
- promoteSingleToNodeList(state)
- }
- is NodeList -> { // LIST -- a list of completion handlers (either new or active)
- if (state.isActive) {
- if (tryMakeCancelling(state, state.list, cause)) return true
- } else {
- // cancelling a non-started coroutine makes it immediately cancelled
- if (updateStateCancelled(state, cause))
- return true
- }
- }
- is Finishing -> { // Completing/Cancelling the job, may cancel
- if (state.cancelled != null) return false // already cancelling
- if (tryMakeCancelling(state, state.list, cause)) return true
- }
- else -> { // is inactive
- return false
- }
- }
- }
- }
-
- // try make expected state in cancelling on the condition that we're still in this state
- private fun tryMakeCancelling(expect: Incomplete, list: NodeList, cause: Throwable?): Boolean {
- val cancelled = Cancelled(this, cause)
- if (!_state.compareAndSet(expect, Finishing(list, cancelled, false))) return false
- onFinishingInternal(cancelled)
- onCancellationInternal(cancelled)
- notifyCancellation(list, cause)
- return true
- }
-
- private fun makeCompletingOnCancel(cause: Throwable?): Boolean =
- makeCompleting(Cancelled(this, cause))
-
- /**
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal fun makeCompleting(proposedUpdate: Any?): Boolean =
- when (makeCompletingInternal(proposedUpdate, mode = MODE_ATOMIC_DEFAULT)) {
- COMPLETING_ALREADY_COMPLETING -> false
- else -> true
- }
-
- /**
- * Returns:
- * * `true` if state was updated to completed/cancelled;
- * * `false` if made completing or it is cancelling and is waiting for children.
- *
- * @throws IllegalStateException if job is already complete or completing
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal actual fun makeCompletingOnce(proposedUpdate: Any?, mode: Int): Boolean =
- when (makeCompletingInternal(proposedUpdate, mode)) {
- COMPLETING_COMPLETED -> true
- COMPLETING_WAITING_CHILDREN -> false
- else -> throw IllegalStateException("Job $this is already complete or completing, " +
- "but is being completed with $proposedUpdate", proposedUpdate.exceptionOrNull)
- }
-
- private fun makeCompletingInternal(proposedUpdate: Any?, mode: Int): Int {
- loopOnState { state ->
- if (state !is Incomplete)
- return COMPLETING_ALREADY_COMPLETING
- if (state is Finishing && state.completing)
- return COMPLETING_ALREADY_COMPLETING
- val child: Child? = firstChild(state) ?: // or else complete immediately w/o children
- when {
- state !is Finishing && hasOnFinishingHandler(proposedUpdate) -> null // unless it has onFinishing handler
- updateState(state, proposedUpdate, mode) -> return COMPLETING_COMPLETED
- else -> return@loopOnState
- }
- val list = state.list ?: // must promote to list to correctly operate on child lists
- when (state) {
- is Empty -> {
- promoteEmptyToNodeList(state)
- return@loopOnState // retry
- }
- is JobNode<*> -> {
- promoteSingleToNodeList(state)
- return@loopOnState // retry
- }
- else -> error("Unexpected state with an empty list: $state")
- }
- // cancel all children in list on exceptional completion
- if (proposedUpdate is CompletedExceptionally)
- child?.cancelChildrenInternal(proposedUpdate.exception)
- // switch to completing state
- val cancelled = (state as? Finishing)?.cancelled ?: (proposedUpdate as? Cancelled)
- val completing = Finishing(list, cancelled, true)
- if (_state.compareAndSet(state, completing)) {
- if (state !is Finishing) onFinishingInternal(proposedUpdate)
- if (child != null && tryWaitForChild(child, proposedUpdate))
- return COMPLETING_WAITING_CHILDREN
- if (updateState(completing, proposedUpdate, mode = MODE_ATOMIC_DEFAULT))
- return COMPLETING_COMPLETED
- }
- }
- }
-
- private tailrec fun Child.cancelChildrenInternal(cause: Throwable) {
- childJob.cancel(JobCancellationException("Child job was cancelled because of parent failure", cause, childJob))
- nextChild()?.cancelChildrenInternal(cause)
- }
-
- private val Any?.exceptionOrNull: Throwable?
- get() = (this as? CompletedExceptionally)?.exception
-
- private fun firstChild(state: Incomplete) =
- state as? Child ?: state.list?.nextChild()
-
- // return false when there is no more incomplete children to wait
- private tailrec fun tryWaitForChild(child: Child, proposedUpdate: Any?): Boolean {
- val handle = child.childJob.invokeOnCompletion(invokeImmediately = false,
- handler = ChildCompletion(this, child, proposedUpdate))
- if (handle !== NonDisposableHandle) return true // child is not complete and we've started waiting for it
- val nextChild = child.nextChild() ?: return false
- return tryWaitForChild(nextChild, proposedUpdate)
- }
-
- /**
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal fun continueCompleting(lastChild: Child, proposedUpdate: Any?) {
- loopOnState { state ->
- if (state !is Finishing)
- throw IllegalStateException("Job $this is found in expected state while completing with $proposedUpdate", proposedUpdate.exceptionOrNull)
- // figure out if we need to wait for next child
- val waitChild = lastChild.nextChild()
- // try wait for next child
- if (waitChild != null && tryWaitForChild(waitChild, proposedUpdate)) return // waiting for next child
- // no more children to wait -- try update state
- if (updateState(state, proposedUpdate, MODE_ATOMIC_DEFAULT)) return
- }
- }
-
- private fun LockFreeLinkedListNode.nextChild(): Child? {
- var cur = this
- while (cur.isRemoved) cur = cur.prev.unwrap() // rollback to prev non-removed (or list head)
- while (true) {
- cur = cur.next.unwrap()
- if (cur.isRemoved) continue
- if (cur is Child) return cur
- if (cur is NodeList) return null // checked all -- no more children
- }
- }
-
- public actual final override val children: Sequence<Job> get() = buildSequence {
- val state = this@JobSupport.state
- when (state) {
- is Child -> yield(state.childJob)
- is Incomplete -> state.list?.let { list ->
- list.forEach<Child> { yield(it.childJob) }
- }
- }
- }
-
- @Suppress("OverridingDeprecatedMember")
- public actual final override fun attachChild(child: Job): DisposableHandle =
- invokeOnCompletion(onCancelling = true, handler = Child(this, child))
-
- @Suppress("OverridingDeprecatedMember")
- public final override fun cancelChildren(cause: Throwable?) {
- this.cancelChildren(cause) // use extension function
- }
-
- /**
- * Override to process any exceptions that were encountered while invoking completion handlers
- * installed via [invokeOnCompletion].
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal actual open fun handleException(exception: Throwable) {
- throw exception
- }
-
- /**
- * This function is invoked once when job is cancelled or is completed, similarly to [invokeOnCompletion] with
- * `onCancelling` set to `true`.
- * @param exceptionally not null when the the job was cancelled or completed exceptionally,
- * null when it has completed normally.
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal actual open fun onCancellationInternal(exceptionally: CompletedExceptionally?) {}
-
- /**
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal actual open fun hasOnFinishingHandler(update: Any?) = false
-
- /**
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal actual open fun onFinishingInternal(update: Any?) {}
-
- /**
- * Override for post-completion actions that need to do something with the state.
- * @param mode completion mode.
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal actual open fun onCompletionInternal(state: Any?, mode: Int) {}
-
- // for nicer debugging
- public override fun toString(): String =
- "${nameString()}{${stateString()}}@$hexAddress"
-
- /**
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal actual open fun nameString(): String = this::class.java.simpleName
-
- private fun stateString(): String {
- val state = this.state
- return when (state) {
- is Finishing -> buildString {
- if (state.cancelled != null) append("Cancelling")
- if (state.completing) append("Completing")
- }
- is Incomplete -> if (state.isActive) "Active" else "New"
- is Cancelled -> "Cancelled"
- is CompletedExceptionally -> "CompletedExceptionally"
- else -> "Completed"
- }
- }
-
- /**
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal interface Incomplete {
- val isActive: Boolean
- val list: NodeList? // is null only for Empty and JobNode incomplete state objects
- }
-
- // Cancelling or Completing
- private class Finishing(
- override val list: NodeList,
- @JvmField val cancelled: Cancelled?, /* != null when cancelling */
- @JvmField val completing: Boolean /* true when completing */
- ) : Incomplete {
- override val isActive: Boolean get() = cancelled == null
- }
-
- private val Incomplete.isCancelling: Boolean
- get() = this is Finishing && cancelled != null
-
- /**
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal class NodeList(
- active: Boolean
- ) : LockFreeLinkedListHead(), Incomplete {
- private val _active = atomic(if (active) 1 else 0)
-
- override val isActive: Boolean get() = _active.value != 0
- override val list: NodeList get() = this
-
- fun tryMakeActive(): Int {
- if (_active.value != 0) return FALSE
- if (_active.compareAndSet(0, 1)) return TRUE
- return RETRY
- }
-
- override fun toString(): String = buildString {
- append("List")
- append(if (isActive) "{Active}" else "{New}")
- append("[")
- var first = true
- this@NodeList.forEach<JobNode<*>> { node ->
- if (first) first = false else append(", ")
- append(node)
- }
- append("]")
- }
- }
-
- /*
- * =================================================================================================
- * This is ready-to-use implementation for Deferred interface.
- * However, it is not type-safe. Conceptually it just exposes the value of the underlying
- * completed state as `Any?`
- * =================================================================================================
- */
-
- public actual val isCompletedExceptionally: Boolean get() = state is CompletedExceptionally
-
- public actual fun getCompletionExceptionOrNull(): Throwable? {
- val state = this.state
- check(state !is Incomplete) { "This job has not completed yet" }
- return state.exceptionOrNull
- }
-
- /**
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal fun getCompletedInternal(): Any? {
- val state = this.state
- check(state !is Incomplete) { "This job has not completed yet" }
- if (state is CompletedExceptionally) throw state.exception
- return state
- }
-
- /**
- * @suppress **This is unstable API and it is subject to change.**
- */
- internal suspend fun awaitInternal(): Any? {
- // fast-path -- check state (avoid extra object creation)
- while(true) { // lock-free loop on state
- val state = this.state
- if (state !is Incomplete) {
- // already complete -- just return result
- if (state is CompletedExceptionally) throw state.exception
- return state
-
- }
- if (startInternal(state) >= 0) break // break unless needs to retry
- }
- return awaitSuspend() // slow-path
- }
-
- private suspend fun awaitSuspend(): Any? = suspendCancellableCoroutine { cont ->
- cont.disposeOnCompletion(invokeOnCompletion {
- val state = this.state
- check(state !is Incomplete)
- if (state is CompletedExceptionally)
- cont.resumeWithException(state.exception)
- else
- cont.resume(state)
- })
- }
-
- /**
- * @suppress **This is unstable API and it is subject to change.**
- */
- // registerSelectAwaitInternal
- @Suppress("UNCHECKED_CAST")
- internal fun <T, R> registerSelectClause1Internal(select: SelectInstance<R>, block: suspend (T) -> R) {
- // fast-path -- check state and select/return if needed
- loopOnState { state ->
- if (select.isSelected) return
- if (state !is Incomplete) {
- // already complete -- select result
- if (select.trySelect(null)) {
- if (state is CompletedExceptionally)
- select.resumeSelectCancellableWithException(state.exception)
- else
- block.startCoroutineUndispatched(state as T, select.completion)
- }
- return
- }
- if (startInternal(state) == 0) {
- // slow-path -- register waiter for completion
- select.disposeOnSelect(invokeOnCompletion(handler = SelectAwaitOnCompletion(this, select, block)))
- return
- }
- }
- }
-
- /**
- * @suppress **This is unstable API and it is subject to change.**
- */
- @Suppress("UNCHECKED_CAST")
- internal fun <T, R> selectAwaitCompletion(select: SelectInstance<R>, block: suspend (T) -> R) {
- val state = this.state
- // Note: await is non-atomic (can be cancelled while dispatched)
- if (state is CompletedExceptionally)
- select.resumeSelectCancellableWithException(state.exception)
- else
- block.startCoroutineCancellable(state as T, select.completion)
- }
-}
-
-internal const val ON_CANCEL_MAKE_CANCELLED = 0
-internal const val ON_CANCEL_MAKE_CANCELLING = 1
-internal const val ON_CANCEL_MAKE_COMPLETING = 2
-
-private const val COMPLETING_ALREADY_COMPLETING = 0
-private const val COMPLETING_COMPLETED = 1
-private const val COMPLETING_WAITING_CHILDREN = 2
-
-private const val RETRY = -1
-private const val FALSE = 0
-private const val TRUE = 1
-
-@Suppress("PrivatePropertyName")
-private val EmptyNew = Empty(false)
-@Suppress("PrivatePropertyName")
-private val EmptyActive = Empty(true)
-
-private class Empty(override val isActive: Boolean) : JobSupport.Incomplete {
- override val list: JobSupport.NodeList? get() = null
- override fun toString(): String = "Empty{${if (isActive) "Active" else "New" }}"
-}
-
-private class JobImpl(parent: Job? = null) : JobSupport(true) {
- init { initParentJobInternal(parent) }
- override val onCancelMode: Int get() = ON_CANCEL_MAKE_COMPLETING
-}
-
-// -------- invokeOnCompletion nodes
-
-internal abstract class JobNode<out J : Job>(
- @JvmField val job: J
-) : LockFreeLinkedListNode(), DisposableHandle, CompletionHandler, JobSupport.Incomplete {
- final override val isActive: Boolean get() = true
- final override val list: JobSupport.NodeList? get() = null
- final override fun dispose() = (job as JobSupport).removeNode(this)
- abstract override fun invoke(reason: Throwable?)
-}
-
-private class InvokeOnCompletion(
- job: Job,
- private val handler: CompletionHandler
-) : JobNode<Job>(job) {
- override fun invoke(reason: Throwable?) = handler.invoke(reason)
- override fun toString() = "InvokeOnCompletion[${handler::class.java.name}@${Integer.toHexString(System.identityHashCode(handler))}]"
-}
-
-private class ResumeOnCompletion(
- job: Job,
- private val continuation: Continuation<Unit>
-) : JobNode<Job>(job) {
- override fun invoke(reason: Throwable?) = continuation.resume(Unit)
- override fun toString() = "ResumeOnCompletion[$continuation]"
-}
-
-internal class DisposeOnCompletion(
- job: Job,
- private val handle: DisposableHandle
-) : JobNode<Job>(job) {
- override fun invoke(reason: Throwable?) = handle.dispose()
- override fun toString(): String = "DisposeOnCompletion[$handle]"
-}
-
-private class CancelFutureOnCompletion(
- job: Job,
- private val future: Future<*>
-) : JobNode<Job>(job) {
- override fun invoke(reason: Throwable?) {
- // Don't interrupt when cancelling future on completion, because no one is going to reset this
- // interruption flag and it will cause spurious failures elsewhere
- future.cancel(false)
- }
- override fun toString() = "CancelFutureOnCompletion[$future]"
-}
-
-private class SelectJoinOnCompletion<R>(
- job: JobSupport,
- private val select: SelectInstance<R>,
- private val block: suspend () -> R
-) : JobNode<JobSupport>(job) {
- override fun invoke(reason: Throwable?) {
- if (select.trySelect(null))
- block.startCoroutineCancellable(select.completion)
- }
- override fun toString(): String = "SelectJoinOnCompletion[$select]"
-}
-
-private class SelectAwaitOnCompletion<T, R>(
- job: JobSupport,
- private val select: SelectInstance<R>,
- private val block: suspend (T) -> R
-) : JobNode<JobSupport>(job) {
- override fun invoke(reason: Throwable?) {
- if (select.trySelect(null))
- job.selectAwaitCompletion(select, block)
- }
- override fun toString(): String = "SelectAwaitOnCompletion[$select]"
-}
-
-// -------- invokeOnCancellation nodes
-
-/**
- * Marker for node that shall be invoked on cancellation (in _cancelling_ state).
- * **Note: may be invoked multiple times during cancellation.**
- */
-internal abstract class JobCancellationNode<out J : Job>(job: J) : JobNode<J>(job)
-
-private class InvokeOnCancellation(
- job: Job,
- private val handler: CompletionHandler
-) : JobCancellationNode<Job>(job) {
- // delegate handler shall be invoked at most once, so here is an additional flag
- private val _invoked = atomic(0)
- override fun invoke(reason: Throwable?) {
- if (_invoked.compareAndSet(0, 1)) handler.invoke(reason)
- }
- override fun toString() = "InvokeOnCancellation[${handler::class.java.name}@${Integer.toHexString(System.identityHashCode(handler))}]"
-}
-
-internal class Child(
- parent: JobSupport,
- @JvmField val childJob: Job
-) : JobCancellationNode<JobSupport>(parent) {
- override fun invoke(reason: Throwable?) {
- // Always materialize the actual instance of parent's completion exception and cancel child with it
- childJob.cancel(job.getCancellationException())
- }
- override fun toString(): String = "Child[$childJob]"
-}
-
-private class ChildCompletion(
- private val parent: JobSupport,
- private val child: Child,
- private val proposedUpdate: Any?
-) : JobNode<Job>(child.childJob) {
- override fun invoke(reason: Throwable?) {
- parent.continueCompleting(child, proposedUpdate)
- }
-}
-
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/NonCancellable.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/NonCancellable.kt
deleted file mode 100644
index 51aa9cf..0000000
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/NonCancellable.kt
+++ /dev/null
@@ -1,91 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental
-
-import kotlinx.coroutines.experimental.NonCancellable.isActive
-import kotlinx.coroutines.experimental.selects.SelectClause0
-import kotlin.coroutines.experimental.AbstractCoroutineContextElement
-
-/**
- * A non-cancelable job that is always [active][isActive]. It is designed for [withContext] function
- * to prevent cancellation of code blocks that need to be executed without cancellation.
- *
- * Use it like this:
- * ```
- * withContext(NonCancellable) {
- * // this code will not be cancelled
- * }
- * ```
- */
-public actual object NonCancellable : AbstractCoroutineContextElement(Job), Job {
- /** Always returns `true`. */
- actual override val isActive: Boolean get() = true
-
- /** Always returns `false`. */
- actual override val isCompleted: Boolean get() = false
-
- /** Always returns `false`. */
- actual override val isCancelled: Boolean get() = false
-
- /** Always returns `false`. */
- actual override fun start(): Boolean = false
-
- /** Always throws [UnsupportedOperationException]. */
- actual suspend override fun join() {
- throw UnsupportedOperationException("This job is always active")
- }
-
- override val onJoin: SelectClause0
- get() = throw UnsupportedOperationException("This job is always active")
-
- /** Always throws [IllegalStateException]. */
- actual override fun getCancellationException(): CancellationException = throw IllegalStateException("This job is always active")
-
- /** Always returns [NonDisposableHandle]. */
- @Suppress("OverridingDeprecatedMember")
- override fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle =
- NonDisposableHandle
-
- /** Always returns [NonDisposableHandle]. */
- @Suppress("OverridingDeprecatedMember")
- override fun invokeOnCompletion(handler: CompletionHandler, onCancelling: Boolean): DisposableHandle =
- NonDisposableHandle
-
- /** Always returns [NonDisposableHandle]. */
- @Suppress("OverridingDeprecatedMember")
- override fun invokeOnCompletion(onCancelling_: Boolean, handler: CompletionHandler): DisposableHandle =
- NonDisposableHandle
-
- /** Always returns [NonDisposableHandle]. */
- actual override fun invokeOnCompletion(onCancelling: Boolean, invokeImmediately: Boolean, handler: CompletionHandler): DisposableHandle =
- NonDisposableHandle
-
- /** Always returns `false`. */
- actual override fun cancel(cause: Throwable?): Boolean = false
-
- /** Always returns [emptySequence]. */
- actual override val children: Sequence<Job>
- get() = emptySequence()
-
- /** Always returns [NonDisposableHandle] and does not do anything. */
- @Suppress("OverridingDeprecatedMember")
- actual override fun attachChild(child: Job): DisposableHandle = NonDisposableHandle
-
- /** Does not do anything. */
- @Suppress("OverridingDeprecatedMember")
- override fun cancelChildren(cause: Throwable?) {}
-}
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/JobTest.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Runnable.kt
similarity index 67%
copy from core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/JobTest.kt
copy to core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Runnable.kt
index 06f7c68..83e239d 100644
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/JobTest.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Runnable.kt
@@ -16,14 +16,14 @@
package kotlinx.coroutines.experimental
-import kotlin.test.*
+/**
+ * A runnable task for [CoroutineDispatcher.dispatch].
+ */
+public actual typealias Runnable = java.lang.Runnable
-class JobTest : TestBase() {
- @Test
- fun testMemoryRelease() {
- val job = Job()
- val n = 10_000_000 * stressTestMultiplier
- var fireCount = 0
- for (i in 0 until n) job.invokeOnCompletion { fireCount++ }.dispose()
- }
-}
\ No newline at end of file
+/**
+ * Creates [Runnable] task instance.
+ */
+@Suppress("FunctionName")
+public actual inline fun Runnable(crossinline block: () -> Unit): Runnable =
+ java.lang.Runnable { block() }
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Scheduled.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Scheduled.kt
deleted file mode 100644
index b881d98..0000000
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Scheduled.kt
+++ /dev/null
@@ -1,184 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental
-
-import kotlinx.coroutines.experimental.intrinsics.*
-import kotlinx.coroutines.experimental.selects.*
-import java.util.concurrent.*
-import kotlin.coroutines.experimental.*
-import kotlin.coroutines.experimental.intrinsics.*
-
-/**
- * Runs a given suspending [block] of code inside a coroutine with a specified timeout and throws
- * [TimeoutCancellationException] if timeout was exceeded.
- *
- * The code that is executing inside the [block] is cancelled on timeout and the active or next invocation of
- * cancellable suspending function inside the block throws [TimeoutCancellationException].
- * Even if the code in the block suppresses [TimeoutCancellationException], it
- * is still thrown by `withTimeout` invocation.
- *
- * The sibling function that does not throw exception on timeout is [withTimeoutOrNull].
- * Note, that timeout action can be specified for [select] invocation with [onTimeout][SelectBuilder.onTimeout] clause.
- *
- * This function delegates to [Delay.invokeOnTimeout] if the context [CoroutineDispatcher]
- * implements [Delay] interface, otherwise it tracks time using a built-in single-threaded scheduled executor service.
- *
- * @param time timeout time in milliseconds.
- */
-public actual suspend fun <T> withTimeout(time: Int, block: suspend CoroutineScope.() -> T): T =
- withTimeout(time.toLong(), TimeUnit.MILLISECONDS, block)
-
-/**
- * Runs a given suspending [block] of code inside a coroutine with a specified timeout and throws
- * [TimeoutCancellationException] if timeout was exceeded.
- *
- * The code that is executing inside the [block] is cancelled on timeout and the active or next invocation of
- * cancellable suspending function inside the block throws [TimeoutCancellationException].
- * Even if the code in the block suppresses [TimeoutCancellationException], it
- * is still thrown by `withTimeout` invocation.
- *
- * The sibling function that does not throw exception on timeout is [withTimeoutOrNull].
- * Note, that timeout action can be specified for [select] invocation with [onTimeout][SelectBuilder.onTimeout] clause.
- *
- * This function delegates to [Delay.invokeOnTimeout] if the context [CoroutineDispatcher]
- * implements [Delay] interface, otherwise it tracks time using a built-in single-threaded scheduled executor service.
- *
- * @param time timeout time
- * @param unit timeout unit (milliseconds by default)
- */
-public suspend fun <T> withTimeout(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS, block: suspend CoroutineScope.() -> T): T {
- require(time >= 0) { "Timeout time $time cannot be negative" }
- if (time <= 0L) throw CancellationException("Timed out immediately")
- return suspendCoroutineOrReturn { cont: Continuation<T> ->
- setupTimeout(TimeoutCoroutine(time, unit, cont), block)
- }
-}
-
-private fun <U, T: U> setupTimeout(
- coroutine: TimeoutCoroutine<U, T>,
- block: suspend CoroutineScope.() -> T
-): Any? {
- // schedule cancellation of this coroutine on time
- val cont = coroutine.cont
- val context = cont.context
- coroutine.disposeOnCompletion(context.delay.invokeOnTimeout(coroutine.time, coroutine.unit, coroutine))
- // restart block using new coroutine with new job,
- // however start it as undispatched coroutine, because we are already in the proper context
- return coroutine.startUndispatchedOrReturn(coroutine, block)
-}
-
-/**
- * @suppress **Deprecated**: for binary compatibility only
- */
-@Deprecated("for binary compatibility only", level=DeprecationLevel.HIDDEN)
-public suspend fun <T> withTimeout(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS, block: suspend () -> T): T =
- withTimeout(time, unit) { block() }
-
-private open class TimeoutCoroutine<U, in T: U>(
- @JvmField val time: Long,
- @JvmField val unit: TimeUnit,
- @JvmField val cont: Continuation<U>
-) : AbstractCoroutine<T>(cont.context, active = true), Runnable, Continuation<T> {
- override val defaultResumeMode: Int get() = MODE_DIRECT
-
- @Suppress("LeakingThis")
- override fun run() {
- cancel(TimeoutCancellationException(time, unit, this))
- }
-
- @Suppress("UNCHECKED_CAST")
- internal override fun onCompletionInternal(state: Any?, mode: Int) {
- if (state is CompletedExceptionally)
- cont.resumeWithExceptionMode(state.exception, mode)
- else
- cont.resumeMode(state as T, mode)
- }
-
- override fun nameString(): String =
- "${super.nameString()}($time $unit)"
-}
-
-/**
- * Runs a given suspending block of code inside a coroutine with a specified timeout and returns
- * `null` if this timeout was exceeded.
- *
- * The code that is executing inside the [block] is cancelled on timeout and the active or next invocation of
- * cancellable suspending function inside the block throws [TimeoutCancellationException].
- * Even if the code in the block suppresses [TimeoutCancellationException], this
- * invocation of `withTimeoutOrNull` still returns `null`.
- *
- * The sibling function that throws exception on timeout is [withTimeout].
- * Note, that timeout action can be specified for [select] invocation with [onTimeout][SelectBuilder.onTimeout] clause.
- *
- * This function delegates to [Delay.invokeOnTimeout] if the context [CoroutineDispatcher]
- * implements [Delay] interface, otherwise it tracks time using a built-in single-threaded scheduled executor service.
- *
- * @param time timeout time in milliseconds.
- */
-public actual suspend fun <T> withTimeoutOrNull(time: Int, block: suspend CoroutineScope.() -> T): T? =
- withTimeoutOrNull(time.toLong(), TimeUnit.MILLISECONDS, block)
-
-/**
- * Runs a given suspending block of code inside a coroutine with a specified timeout and returns
- * `null` if this timeout was exceeded.
- *
- * The code that is executing inside the [block] is cancelled on timeout and the active or next invocation of
- * cancellable suspending function inside the block throws [TimeoutCancellationException].
- * Even if the code in the block suppresses [TimeoutCancellationException], this
- * invocation of `withTimeoutOrNull` still returns `null`.
- *
- * The sibling function that throws exception on timeout is [withTimeout].
- * Note, that timeout action can be specified for [select] invocation with [onTimeout][SelectBuilder.onTimeout] clause.
- *
- * This function delegates to [Delay.invokeOnTimeout] if the context [CoroutineDispatcher]
- * implements [Delay] interface, otherwise it tracks time using a built-in single-threaded scheduled executor service.
- *
- * @param time timeout time
- * @param unit timeout unit (milliseconds by default)
- */
-public suspend fun <T> withTimeoutOrNull(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS, block: suspend CoroutineScope.() -> T): T? {
- require(time >= 0) { "Timeout time $time cannot be negative" }
- if (time <= 0L) return null
- return suspendCoroutineOrReturn { cont: Continuation<T?> ->
- setupTimeout(TimeoutOrNullCoroutine(time, unit, cont), block)
- }
-}
-
-/**
- * @suppress **Deprecated**: for binary compatibility only
- */
-@Deprecated("for binary compatibility only", level=DeprecationLevel.HIDDEN)
-public suspend fun <T> withTimeoutOrNull(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS, block: suspend () -> T): T? =
- withTimeoutOrNull(time, unit) { block() }
-
-private class TimeoutOrNullCoroutine<T>(
- time: Long,
- unit: TimeUnit,
- cont: Continuation<T?>
-) : TimeoutCoroutine<T?, T>(time, unit, cont) {
- @Suppress("UNCHECKED_CAST")
- internal override fun onCompletionInternal(state: Any?, mode: Int) {
- if (state is CompletedExceptionally) {
- val exception = state.exception
- if (exception is TimeoutCancellationException && exception.coroutine === this)
- cont.resumeMode(null, mode) else
- cont.resumeWithExceptionMode(exception, mode)
- } else
- cont.resumeMode(state as T, mode)
- }
-}
-
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Yield.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Yield.kt
deleted file mode 100644
index 833a728..0000000
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Yield.kt
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental
-
-import kotlin.coroutines.experimental.CoroutineContext
-import kotlin.coroutines.experimental.intrinsics.COROUTINE_SUSPENDED
-import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
-
-/**
- * Yields a thread (or thread pool) of the current coroutine dispatcher to other coroutines to run.
- * If the coroutine dispatcher does not have its own thread pool (like [Unconfined] dispatcher) then this
- * function does nothing, but checks if the coroutine [Job] was completed.
- * This suspending function is cancellable.
- * If the [Job] of the current coroutine is cancelled or completed when this suspending function is invoked or while
- * this function is waiting for dispatching, it resumes with [CancellationException].
- */
-public actual suspend fun yield(): Unit = suspendCoroutineOrReturn sc@ { cont ->
- val context = cont.context
- context.checkCompletion()
- if (cont !is DispatchedContinuation<Unit>) return@sc Unit
- if (!cont.dispatcher.isDispatchNeeded(context)) return@sc Unit
- cont.dispatchYield(Unit)
- COROUTINE_SUSPENDED
-}
-
-internal fun CoroutineContext.checkCompletion() {
- val job = get(Job)
- if (job != null && !job.isActive) throw job.getCancellationException()
-}
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/AbstractChannel.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/AbstractChannel.kt
index a42517a..4ce28da 100644
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/AbstractChannel.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/AbstractChannel.kt
@@ -138,7 +138,7 @@
*/
protected fun describeSendBuffered(element: E): AddLastDesc<*> = SendBufferedDesc(queue, element)
- private open class SendBufferedDesc<out E>(
+ private open class SendBufferedDesc<E>(
queue: LockFreeLinkedListHead,
element: E
) : AddLastDesc<SendBuffered<E>>(queue, SendBuffered(element)) {
@@ -153,7 +153,7 @@
*/
protected fun describeSendConflated(element: E): AddLastDesc<*> = SendConflatedDesc(queue, element)
- private class SendConflatedDesc<out E>(
+ private class SendConflatedDesc<E>(
queue: LockFreeLinkedListHead,
element: E
) : SendBufferedDesc<E>(queue, element) {
@@ -647,7 +647,7 @@
}
}
- private inner class TryEnqueueReceiveDesc<in E, R>(
+ private inner class TryEnqueueReceiveDesc<E, R>(
select: SelectInstance<R>,
block: suspend (E?) -> R,
nullOnClose: Boolean
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Actor.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Actor.kt
index d10bcf6..54adc81 100644
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Actor.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Actor.kt
@@ -172,7 +172,8 @@
parentContext: CoroutineContext,
channel: Channel<E>,
private val block: suspend ActorScope<E>.() -> Unit
-) : ActorCoroutine<E>(parentContext, channel, active = false), SelectClause2<E, SendChannel<E>> {
+) : ActorCoroutine<E>(parentContext, channel, active = false),
+ SelectClause2<E, SendChannel<E>> {
override fun onStart() {
block.startCoroutineCancellable(this, this)
}
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/internal/Atomic.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/internal/Atomic.kt
deleted file mode 100644
index 1eaa6e6..0000000
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/internal/Atomic.kt
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental.internal
-
-import kotlinx.atomicfu.atomic
-
-/**
- * The most abstract operation that can be in process. Other threads observing an instance of this
- * class in the fields of their object shall invoke [perform] to help.
- *
- * @suppress **This is unstable API and it is subject to change.**
- */
-public abstract class OpDescriptor {
- /**
- * Returns `null` is operation was performed successfully or some other
- * object that indicates the failure reason.
- */
- abstract fun perform(affected: Any?): Any?
-}
-
-private val NO_DECISION: Any = Symbol("NO_DECISION")
-
-/**
- * Descriptor for multi-word atomic operation.
- * Based on paper
- * ["A Practical Multi-Word Compare-and-Swap Operation"](http://www.cl.cam.ac.uk/research/srgnetos/papers/2002-casn.pdf)
- * by Timothy L. Harris, Keir Fraser and Ian A. Pratt.
- *
- * Note: parts of atomic operation must be globally ordered. Otherwise, this implementation will produce
- * [StackOverflowError].
- *
- * @suppress **This is unstable API and it is subject to change.**
- */
-public abstract class AtomicOp<in T> : OpDescriptor() {
- private val _consensus = atomic<Any?>(NO_DECISION)
-
- val isDecided: Boolean get() = _consensus.value !== NO_DECISION
-
- fun tryDecide(decision: Any?): Boolean {
- check(decision !== NO_DECISION)
- return _consensus.compareAndSet(NO_DECISION, decision)
- }
-
- private fun decide(decision: Any?): Any? = if (tryDecide(decision)) decision else _consensus.value
-
- abstract fun prepare(affected: T): Any? // `null` if Ok, or failure reason
-
- abstract fun complete(affected: T, failure: Any?) // failure != null if failed to prepare op
-
- // returns `null` on success
- @Suppress("UNCHECKED_CAST")
- final override fun perform(affected: Any?): Any? {
- // make decision on status
- var decision = this._consensus.value
- if (decision === NO_DECISION)
- decision = decide(prepare(affected as T))
- complete(affected as T, decision)
- return decision
- }
-}
-
-/**
- * A part of multi-step atomic operation [AtomicOp].
- *
- * @suppress **This is unstable API and it is subject to change.**
- */
-public abstract class AtomicDesc {
- abstract fun prepare(op: AtomicOp<*>): Any? // returns `null` if prepared successfully
- abstract fun complete(op: AtomicOp<*>, failure: Any?) // decision == null if success
-}
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/internal/LockFreeLinkedList.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/internal/LockFreeLinkedList.kt
index 06e9384..17a0ebb 100644
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/internal/LockFreeLinkedList.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/internal/LockFreeLinkedList.kt
@@ -41,15 +41,14 @@
private val REMOVE_PREPARED: Any = Symbol("REMOVE_PREPARED")
-/**
- * @suppress **This is unstable API and it is subject to change.**
- */
+/** @suppress **This is unstable API and it is subject to change.** */
public typealias RemoveFirstDesc<T> = LockFreeLinkedListNode.RemoveFirstDesc<T>
-/**
- * @suppress **This is unstable API and it is subject to change.**
- */
-public typealias AddLastDesc<T> = LockFreeLinkedListNode.AddLastDesc<T>
+/** @suppress **This is unstable API and it is subject to change.** */
+public actual typealias AddLastDesc<T> = LockFreeLinkedListNode.AddLastDesc<T>
+
+/** @suppress **This is unstable API and it is subject to change.** */
+public actual typealias AbstractAtomicDesc = LockFreeLinkedListNode.AbstractAtomicDesc
/**
* Doubly-linked concurrent list node with remove support.
@@ -67,7 +66,7 @@
* @suppress **This is unstable API and it is subject to change.**
*/
@Suppress("LeakingThis")
-public open class LockFreeLinkedListNode {
+public actual open class LockFreeLinkedListNode {
private val _next = atomic<Any>(this) // Node | Removed | OpDescriptor
private val _prev = atomic<Any>(this) // Node | Removed
private val _removedRef = atomic<Removed?>(null) // lazily cached removed ref to this
@@ -99,7 +98,7 @@
public val isFresh: Boolean get() = _next.value === this
- public val isRemoved: Boolean get() = next is Removed
+ public actual val isRemoved: Boolean get() = next is Removed
// LINEARIZABLE. Returns Node | Removed
public val next: Any get() {
@@ -109,6 +108,8 @@
}
}
+ public actual val nextNode: Node get() = next.unwrap()
+
// LINEARIZABLE. Returns Node | Removed
public val prev: Any get() {
_prev.loop { prev ->
@@ -119,9 +120,11 @@
}
}
+ public actual val prevNode: Node get() = prev.unwrap()
+
// ------ addOneIfEmpty ------
- public fun addOneIfEmpty(node: Node): Boolean {
+ public actual fun addOneIfEmpty(node: Node): Boolean {
node._prev.lazySet(this)
node._next.lazySet(this)
while (true) {
@@ -140,7 +143,7 @@
/**
* Adds last item to this list.
*/
- public fun addLast(node: Node) {
+ public actual fun addLast(node: Node) {
while (true) { // lock-free loop on prev.next
val prev = prev as Node // sentinel node is never removed, so prev is always defined
if (prev.addNext(node, this)) return
@@ -152,7 +155,7 @@
/**
* Adds last item to this list atomically if the [condition] is true.
*/
- public inline fun addLastIf(node: Node, crossinline condition: () -> Boolean): Boolean {
+ public actual inline fun addLastIf(node: Node, crossinline condition: () -> Boolean): Boolean {
val condAdd = makeCondAddOp(node, condition)
while (true) { // lock-free loop on prev.next
val prev = prev as Node // sentinel node is never removed, so prev is always defined
@@ -163,7 +166,7 @@
}
}
- public inline fun addLastIfPrev(node: Node, predicate: (Node) -> Boolean): Boolean {
+ public actual inline fun addLastIfPrev(node: Node, predicate: (Node) -> Boolean): Boolean {
while (true) { // lock-free loop on prev.next
val prev = prev as Node // sentinel node is never removed, so prev is always defined
if (!predicate(prev)) return false
@@ -171,7 +174,7 @@
}
}
- public inline fun addLastIfPrevAndIf(
+ public actual inline fun addLastIfPrevAndIf(
node: Node,
predicate: (Node) -> Boolean, // prev node predicate
crossinline condition: () -> Boolean // atomically checked condition
@@ -239,7 +242,7 @@
* Removes this node from the list. Returns `true` when removed successfully, or `false` if the node was already
* removed or if it was not added to any list in the first place.
*/
- public open fun remove(): Boolean {
+ public actual open fun remove(): Boolean {
while (true) { // lock-free loop on next
val next = this.next
if (next is Removed) return false // was already removed -- don't try to help (original thread will take care)
@@ -273,7 +276,7 @@
}
}
- public fun removeFirstOrNull(): Node? {
+ public actual fun removeFirstOrNull(): Node? {
while (true) { // try to linearize
val first = next as Node
if (first === this) return null
@@ -295,7 +298,7 @@
}
// just peek at item when predicate is true
- public inline fun <reified T> removeFirstIfIsInstanceOfOrPeekIf(predicate: (T) -> Boolean): T? {
+ public actual inline fun <reified T> removeFirstIfIsInstanceOfOrPeekIf(predicate: (T) -> Boolean): T? {
while (true) { // try to linearize
val first = next as Node
if (first === this) return null
@@ -308,7 +311,7 @@
// ------ multi-word atomic operations helpers ------
- public open class AddLastDesc<out T : Node>(
+ public open class AddLastDesc<T : Node> actual constructor(
@JvmField val queue: Node,
@JvmField val node: T
) : AbstractAtomicDesc() {
@@ -340,7 +343,7 @@
override fun retry(affected: Node, next: Any): Boolean = next !== queue
- override fun onPrepare(affected: Node, next: Node): Any? {
+ protected override fun onPrepare(affected: Node, next: Node): Any? {
// Note: onPrepare must use CAS to make sure the stale invocation is not
// going to overwrite the previous decision on successful preparation.
// Result of CAS is irrelevant, but we must ensure that it is set when invoker completes
@@ -659,20 +662,20 @@
}
@PublishedApi
-internal fun Any.unwrap(): Node = if (this is Removed) ref else this as Node
+internal fun Any.unwrap(): Node = (this as? Removed)?.ref ?: this as Node
/**
* Head (sentinel) item of the linked list that is never removed.
*
* @suppress **This is unstable API and it is subject to change.**
*/
-public open class LockFreeLinkedListHead : LockFreeLinkedListNode() {
- public val isEmpty: Boolean get() = next === this
+public actual open class LockFreeLinkedListHead : LockFreeLinkedListNode() {
+ public actual val isEmpty: Boolean get() = next === this
/**
* Iterates over all elements in this list of a specified type.
*/
- public inline fun <reified T : Node> forEach(block: (T) -> Unit) {
+ public actual inline fun <reified T : Node> forEach(block: (T) -> Unit) {
var cur: Node = next as Node
while (cur != this) {
if (cur is T) block(cur)
@@ -681,8 +684,9 @@
}
// just a defensive programming -- makes sure that list head sentinel is never removed
- public final override fun remove() = throw UnsupportedOperationException()
- public final override fun describeRemove(): AtomicDesc? = throw UnsupportedOperationException()
+ public actual final override fun remove(): Nothing = throw UnsupportedOperationException()
+
+ public final override fun describeRemove(): Nothing = throw UnsupportedOperationException()
internal fun validate() {
var prev: Node = this
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/selects/Select.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/selects/Select.kt
deleted file mode 100644
index 8c6b208..0000000
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/selects/Select.kt
+++ /dev/null
@@ -1,462 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental.selects
-
-import kotlinx.atomicfu.*
-import kotlinx.coroutines.experimental.*
-import kotlinx.coroutines.experimental.channels.*
-import kotlinx.coroutines.experimental.internal.*
-import kotlinx.coroutines.experimental.intrinsics.*
-import kotlinx.coroutines.experimental.sync.*
-import java.util.concurrent.*
-import kotlin.coroutines.experimental.*
-import kotlin.coroutines.experimental.intrinsics.*
-
-/**
- * Scope for [select] invocation.
- */
-public interface SelectBuilder<in R> {
- /**
- * Registers clause in this [select] expression without additional parameters that does not select any value.
- */
- public operator fun SelectClause0.invoke(block: suspend () -> R)
-
- /**
- * Registers clause in this [select] expression without additional parameters that selects value of type [Q].
- */
- public operator fun <Q> SelectClause1<Q>.invoke(block: suspend (Q) -> R)
-
- /**
- * Registers clause in this [select] expression with additional parameter of type [P] that selects value of type [Q].
- */
- public operator fun <P, Q> SelectClause2<P, Q>.invoke(param: P, block: suspend (Q) -> R)
-
- /**
- * Registers clause in this [select] expression with additional parameter nullable parameter of type [P]
- * with the `null` value for this parameter that selects value of type [Q].
- */
- public operator fun <P, Q> SelectClause2<P?, Q>.invoke(block: suspend (Q) -> R) = invoke(null, block)
-
- /**
- * Clause that selects the given [block] after a specified timeout passes.
- *
- * @param time timeout time
- * @param unit timeout unit (milliseconds by default)
- */
- public fun onTimeout(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS, block: suspend () -> R)
-
- /** @suppress **Deprecated: for binary compatibility only **/
- @Deprecated("for binary compatibility only", level=DeprecationLevel.HIDDEN)
- public fun Job.onJoin(block: suspend () -> R) { onJoin(block) }
-
- /** @suppress **Deprecated: for binary compatibility only **/
- @Deprecated("for binary compatibility only", level=DeprecationLevel.HIDDEN)
- public fun <T> Deferred<T>.onAwait(block: suspend (T) -> R) { onAwait(block) }
-
- /** @suppress **Deprecated: for binary compatibility only **/
- @Deprecated("for binary compatibility only", level=DeprecationLevel.HIDDEN)
- public fun Mutex.onLock(owner: Any? = null, block: suspend () -> R) { onLock { block() } }
-
- /** @suppress **Deprecated: for binary compatibility only **/
- @Deprecated("for binary compatibility only", level=DeprecationLevel.HIDDEN)
- public fun <E> SendChannel<E>.onSend(element: E, block: suspend () -> R) { onSend(element) { block() } }
-
- /** @suppress **Deprecated: for binary compatibility only **/
- @Deprecated("for binary compatibility only", level=DeprecationLevel.HIDDEN)
- public fun <E> ReceiveChannel<E>.onReceive(block: suspend (E) -> R) { onReceive(block) }
-
- /** @suppress **Deprecated: for binary compatibility only **/
- @Deprecated("for binary compatibility only", level=DeprecationLevel.HIDDEN)
- public fun <E> ReceiveChannel<E>.onReceiveOrNull(block: suspend (E?) -> R) { onReceiveOrNull(block) }
-}
-
-/**
- * Clause for [select] expression without additional parameters that does not select any value.
- */
-public interface SelectClause0 {
- /**
- * Registers this clause with the specified [select] instance and [block] of code.
- * @suppress **This is unstable API and it is subject to change.**
- */
- public fun <R> registerSelectClause0(select: SelectInstance<R>, block: suspend () -> R)
-}
-
-/**
- * Clause for [select] expression without additional parameters that selects value of type [Q].
- */
-public interface SelectClause1<out Q> {
- /**
- * Registers this clause with the specified [select] instance and [block] of code.
- * @suppress **This is unstable API and it is subject to change.**
- */
- public fun <R> registerSelectClause1(select: SelectInstance<R>, block: suspend (Q) -> R)
-}
-
-/**
- * Clause for [select] expression with additional parameter of type [P] that selects value of type [Q].
- */
-public interface SelectClause2<in P, out Q> {
- /**
- * Registers this clause with the specified [select] instance and [block] of code.
- * @suppress **This is unstable API and it is subject to change.**
- */
- public fun <R> registerSelectClause2(select: SelectInstance<R>, param: P, block: suspend (Q) -> R)
-}
-
-/**
- * Internal representation of select instance. This instance is called _selected_ when
- * the clause to execute is already picked.
- *
- * @suppress **This is unstable API and it is subject to change.**
- */
-public interface SelectInstance<in R> {
- /**
- * Returns `true` when this [select] statement had already picked a clause to execute.
- */
- public val isSelected: Boolean
-
- /**
- * Tries to select this instance.
- */
- public fun trySelect(idempotent: Any?): Boolean
-
- /**
- * Performs action atomically with [trySelect].
- */
- public fun performAtomicTrySelect(desc: AtomicDesc): Any?
-
- /**
- * Performs action atomically when [isSelected] is `false`.
- */
- public fun performAtomicIfNotSelected(desc: AtomicDesc): Any?
-
- /**
- * Returns completion continuation of this select instance.
- * This select instance must be _selected_ first.
- * All resumption through this instance happen _directly_ without going through dispatcher ([MODE_DIRECT]).
- */
- public val completion: Continuation<R>
-
- /**
- * Resumes this instance in a cancellable way ([MODE_CANCELLABLE]).
- */
- public fun resumeSelectCancellableWithException(exception: Throwable)
-
- /**
- * Disposes the specified handle when this instance is selected.
- */
- public fun disposeOnSelect(handle: DisposableHandle)
-}
-
-/**
- * Waits for the result of multiple suspending functions simultaneously, which are specified using _clauses_
- * in the [builder] scope of this select invocation. The caller is suspended until one of the clauses
- * is either _selected_ or _fails_.
- *
- * At most one clause is *atomically* selected and its block is executed. The result of the selected clause
- * becomes the result of the select. If any clause _fails_, then the select invocation produces the
- * corresponding exception. No clause is selected in this case.
- *
- * This select function is _biased_ to the first clause. When multiple clauses can be selected at the same time,
- * the first one of them gets priority. Use [selectUnbiased] for an unbiased (randomized) selection among
- * the clauses.
-
- * There is no `default` clause for select expression. Instead, each selectable suspending function has the
- * corresponding non-suspending version that can be used with a regular `when` expression to select one
- * of the alternatives or to perform default (`else`) action if none of them can be immediately selected.
- *
- * | **Receiver** | **Suspending function** | **Select clause** | **Non-suspending version**
- * | ---------------- | --------------------------------------------- | ------------------------------------------------ | --------------------------
- * | [Job] | [join][Job.join] | [onJoin][SelectBuilder.onJoin] | [isCompleted][Job.isCompleted]
- * | [Deferred] | [await][Deferred.await] | [onAwait][SelectBuilder.onAwait] | [isCompleted][Job.isCompleted]
- * | [SendChannel] | [send][SendChannel.send] | [onSend][SelectBuilder.onSend] | [offer][SendChannel.offer]
- * | [ReceiveChannel] | [receive][ReceiveChannel.receive] | [onReceive][SelectBuilder.onReceive] | [poll][ReceiveChannel.poll]
- * | [ReceiveChannel] | [receiveOrNull][ReceiveChannel.receiveOrNull] | [onReceiveOrNull][SelectBuilder.onReceiveOrNull] | [poll][ReceiveChannel.poll]
- * | [Mutex] | [lock][Mutex.lock] | [onLock][SelectBuilder.onLock] | [tryLock][Mutex.tryLock]
- * | none | [delay] | [onTimeout][SelectBuilder.onTimeout] | none
- *
- * 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][SelectBuilder.onSend], [onReceive][SelectBuilder.onReceive],
- * [onReceiveOrNull][SelectBuilder.onReceiveOrNull], and [onLock][SelectBuilder.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.
- */
-public suspend inline fun <R> select(crossinline builder: SelectBuilder<R>.() -> Unit): R =
- suspendCoroutineOrReturn { cont ->
- val scope = SelectBuilderImpl(cont)
- try {
- builder(scope)
- } catch (e: Throwable) {
- scope.handleBuilderException(e)
- }
- scope.getResult()
- }
-
-
-internal val ALREADY_SELECTED: Any = Symbol("ALREADY_SELECTED")
-private val UNDECIDED: Any = Symbol("UNDECIDED")
-private val RESUMED: Any = Symbol("RESUMED")
-
-@PublishedApi
-internal class SelectBuilderImpl<in R>(
- private val delegate: Continuation<R>
-) : LockFreeLinkedListHead(), SelectBuilder<R>, SelectInstance<R>, Continuation<R> {
- // selection state is "this" (list of nodes) initially and is replaced by idempotent marker (or null) when selected
- private val _state = atomic<Any?>(this)
-
- // this is basically our own SafeContinuation
- private val _result = atomic<Any?>(UNDECIDED)
-
- // cancellability support
- @Volatile
- private var parentHandle: DisposableHandle? = null
-
- /* Result state machine
-
- +-----------+ getResult +---------------------+ resume +---------+
- | UNDECIDED | ------------> | COROUTINE_SUSPENDED | ---------> | RESUMED |
- +-----------+ +---------------------+ +---------+
- |
- | resume
- V
- +------------+ getResult
- | value/Fail | -----------+
- +------------+ |
- ^ |
- | |
- +-------------------+
- */
-
- override val context: CoroutineContext get() = delegate.context
-
- override val completion: Continuation<R> get() = this
-
- private inline fun doResume(value: () -> Any?, block: () -> Unit) {
- check(isSelected) { "Must be selected first" }
- _result.loop { result ->
- when {
- result === UNDECIDED -> if (_result.compareAndSet(UNDECIDED, value())) return
- result === COROUTINE_SUSPENDED -> if (_result.compareAndSet(COROUTINE_SUSPENDED, RESUMED)) {
- block()
- return
- }
- else -> throw IllegalStateException("Already resumed")
- }
- }
- }
-
- // Resumes in MODE_DIRECT
- override fun resume(value: R) {
- doResume({ value }) {
- delegate.resumeDirect(value)
- }
- }
-
- // Resumes in MODE_DIRECT
- override fun resumeWithException(exception: Throwable) {
- doResume({ Fail(exception) }) {
- delegate.resumeDirectWithException(exception)
- }
- }
-
- // Resumes in MODE_CANCELLABLE
- override fun resumeSelectCancellableWithException(exception: Throwable) {
- doResume({ Fail(exception) }) {
- delegate.resumeCancellableWithException(exception)
- }
- }
-
- @PublishedApi
- internal fun getResult(): Any? {
- if (!isSelected) initCancellability()
- var result = _result.value // atomic read
- if (result === UNDECIDED) {
- if (_result.compareAndSet(UNDECIDED, COROUTINE_SUSPENDED)) return COROUTINE_SUSPENDED
- result = _result.value // reread volatile var
- }
- when {
- result === RESUMED -> throw IllegalStateException("Already resumed")
- result is Fail -> throw result.exception
- else -> return result // either COROUTINE_SUSPENDED or data
- }
- }
-
- private fun initCancellability() {
- val parent = context[Job] ?: return
- val newRegistration = parent.invokeOnCompletion(onCancelling = true, handler = SelectOnCancellation(parent))
- parentHandle = newRegistration
- // now check our state _after_ registering
- if (isSelected) newRegistration.dispose()
- }
-
- private inner class SelectOnCancellation(job: Job) : JobCancellationNode<Job>(job) {
- // Note: may be invoked multiple times, but only the first trySelect succeeds anyway
- override fun invoke(reason: Throwable?) {
- if (trySelect(null))
- resumeSelectCancellableWithException(job.getCancellationException())
- }
- override fun toString(): String = "SelectOnCancellation[${this@SelectBuilderImpl}]"
- }
-
- private val state: Any? get() {
- _state.loop { state ->
- if (state !is OpDescriptor) return state
- state.perform(this)
- }
- }
-
- @PublishedApi
- internal fun handleBuilderException(e: Throwable) {
- if (trySelect(null))
- resumeWithException(e)
- else
- handleCoroutineException(context, e)
- }
-
- override val isSelected: Boolean get() = state !== this
-
- override fun disposeOnSelect(handle: DisposableHandle) {
- val node = DisposeNode(handle)
- while (true) { // lock-free loop on state
- val state = this.state
- if (state === this) {
- if (addLastIf(node, { this.state === this }))
- return
- } else { // already selected
- handle.dispose()
- return
- }
- }
- }
-
- private fun doAfterSelect() {
- parentHandle?.dispose()
- forEach<DisposeNode> {
- it.handle.dispose()
- }
- }
-
- // it is just like start(), but support idempotent start
- override fun trySelect(idempotent: Any?): Boolean {
- check(idempotent !is OpDescriptor) { "cannot use OpDescriptor as idempotent marker"}
- while (true) { // lock-free loop on state
- val state = this.state
- when {
- state === this -> {
- if (_state.compareAndSet(this, idempotent)) {
- doAfterSelect()
- return true
- }
- }
- // otherwise -- already selected
- idempotent == null -> return false // already selected
- state === idempotent -> return true // was selected with this marker
- else -> return false
- }
- }
- }
-
- override fun performAtomicTrySelect(desc: AtomicDesc): Any? = AtomicSelectOp(desc, true).perform(null)
- override fun performAtomicIfNotSelected(desc: AtomicDesc): Any? = AtomicSelectOp(desc, false).perform(null)
-
- private inner class AtomicSelectOp(
- @JvmField val desc: AtomicDesc,
- @JvmField val select: Boolean
- ) : AtomicOp<Any?>() {
- override fun prepare(affected: Any?): Any? {
- // only originator of operation makes preparation move of installing descriptor into this selector's state
- // helpers should never do it, or risk ruining progress when they come late
- if (affected == null) {
- // we are originator (affected reference is not null if helping)
- prepareIfNotSelected()?.let { return it }
- }
- return desc.prepare(this)
- }
-
- override fun complete(affected: Any?, failure: Any?) {
- completeSelect(failure)
- desc.complete(this, failure)
- }
-
- fun prepareIfNotSelected(): Any? {
- _state.loop { state ->
- when {
- state === this@AtomicSelectOp -> return null // already in progress
- state is OpDescriptor -> state.perform(this@SelectBuilderImpl) // help
- state === this@SelectBuilderImpl -> {
- if (_state.compareAndSet(this@SelectBuilderImpl, this@AtomicSelectOp))
- return null // success
- }
- else -> return ALREADY_SELECTED
- }
- }
- }
-
- private fun completeSelect(failure: Any?) {
- val selectSuccess = select && failure == null
- val update = if (selectSuccess) null else this@SelectBuilderImpl
- if (_state.compareAndSet(this@AtomicSelectOp, update)) {
- if (selectSuccess)
- doAfterSelect()
- }
- }
- }
-
- override fun SelectClause0.invoke(block: suspend () -> R) {
- registerSelectClause0(this@SelectBuilderImpl, block)
- }
-
- override fun <Q> SelectClause1<Q>.invoke(block: suspend (Q) -> R) {
- registerSelectClause1(this@SelectBuilderImpl, block)
- }
-
- override fun <P, Q> SelectClause2<P, Q>.invoke(param: P, block: suspend (Q) -> R) {
- registerSelectClause2(this@SelectBuilderImpl, param, block)
- }
-
- override fun onTimeout(time: Long, unit: TimeUnit, block: suspend () -> R) {
- require(time >= 0) { "Timeout time $time cannot be negative" }
- if (time == 0L) {
- if (trySelect(null))
- block.startCoroutineUndispatched(completion)
- return
- }
- val action = Runnable {
- // todo: we could have replaced startCoroutine with startCoroutineUndispatched
- // But we need a way to know that Delay.invokeOnTimeout had used the right thread
- if (trySelect(null))
- block.startCoroutineCancellable(completion) // shall be cancellable while waits for dispatch
- }
- disposeOnSelect(context.delay.invokeOnTimeout(time, unit, action))
- }
-
- private class DisposeNode(
- @JvmField val handle: DisposableHandle
- ) : LockFreeLinkedListNode()
-
- private class Fail(
- @JvmField val exception: Throwable
- )
-}
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/selects/SelectUnbiased.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/selects/SelectUnbiased.kt
deleted file mode 100644
index cf79b47..0000000
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/selects/SelectUnbiased.kt
+++ /dev/null
@@ -1,82 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental.selects
-
-import java.util.*
-import java.util.concurrent.TimeUnit
-import kotlin.coroutines.experimental.Continuation
-import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
-
-/**
- * Waits for the result of multiple suspending functions simultaneously like [select], but in an _unbiased_
- * way when multiple clauses are selectable at the same time.
- *
- * This unbiased implementation of `select` expression randomly shuffles the clauses before checking
- * if they are selectable, thus ensuring that there is no statistical bias to the selection of the first
- * clauses.
- *
- * See [select] function description for all the other details.
- */
-public inline suspend fun <R> selectUnbiased(crossinline builder: SelectBuilder<R>.() -> Unit): R =
- suspendCoroutineOrReturn { cont ->
- val scope = UnbiasedSelectBuilderImpl(cont)
- try {
- builder(scope)
- } catch (e: Throwable) {
- scope.handleBuilderException(e)
- }
- scope.initSelectResult()
- }
-
-
-@PublishedApi
-internal class UnbiasedSelectBuilderImpl<in R>(cont: Continuation<R>) : SelectBuilder<R> {
- val instance = SelectBuilderImpl(cont)
- val clauses = arrayListOf<() -> Unit>()
-
- @PublishedApi
- internal fun handleBuilderException(e: Throwable) = instance.handleBuilderException(e)
-
- @PublishedApi
- internal fun initSelectResult(): Any? {
- if (!instance.isSelected) {
- try {
- Collections.shuffle(clauses)
- clauses.forEach { it.invoke() }
- } catch (e: Throwable) {
- instance.handleBuilderException(e)
- }
- }
- return instance.getResult()
- }
-
- override fun SelectClause0.invoke(block: suspend () -> R) {
- clauses += { registerSelectClause0(instance, block) }
- }
-
- override fun <Q> SelectClause1<Q>.invoke(block: suspend (Q) -> R) {
- clauses += { registerSelectClause1(instance, block) }
- }
-
- override fun <P, Q> SelectClause2<P, Q>.invoke(param: P, block: suspend (Q) -> R) {
- clauses += { registerSelectClause2(instance, param, block) }
- }
-
- override fun onTimeout(time: Long, unit: TimeUnit, block: suspend () -> R) {
- clauses += { instance.onTimeout(time, unit, block) }
- }
-}
\ No newline at end of file
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/selects/WhileSelect.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/selects/WhileSelect.kt
deleted file mode 100644
index 0548685..0000000
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/selects/WhileSelect.kt
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental.selects
-
-/**
- * Loops while [select] expression returns `true`.
- *
- * The statement of the form:
- *
- * ```
- * whileSelect {
- * /*body*/
- * }
- * ```
- *
- * is a shortcut for:
- *
- * ```
- * while(select<Boolean> {
- * /*body*/
- * }) {}
- */
-suspend fun whileSelect(builder: SelectBuilder<Boolean>.() -> Unit) {
- while(select<Boolean>(builder)) {}
-}
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/sync/Mutex.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/sync/Mutex.kt
deleted file mode 100644
index 3130085..0000000
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/sync/Mutex.kt
+++ /dev/null
@@ -1,499 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental.sync
-
-import kotlinx.atomicfu.atomic
-import kotlinx.atomicfu.loop
-import kotlinx.coroutines.experimental.*
-import kotlinx.coroutines.experimental.internal.*
-import kotlinx.coroutines.experimental.intrinsics.startCoroutineUndispatched
-import kotlinx.coroutines.experimental.selects.ALREADY_SELECTED
-import kotlinx.coroutines.experimental.selects.SelectClause2
-import kotlinx.coroutines.experimental.selects.SelectInstance
-import kotlinx.coroutines.experimental.selects.select
-import kotlin.coroutines.experimental.startCoroutine
-
-/**
- * Mutual exclusion for coroutines.
- *
- * Mutex has two states: _locked_ and _unlocked_.
- * It is **non-reentrant**, that is invoking [lock] even from the same thread/coroutine that currently holds
- * the lock still suspends the invoker.
- */
-public interface Mutex {
- /**
- * Factory for [Mutex] instances.
- * @suppress **Deprecated**
- */
- public companion object Factory {
- /**
- * Creates new [Mutex] instance.
- * @suppress **Deprecated**
- */
- @Deprecated("Replaced with top-level function", level = DeprecationLevel.HIDDEN)
- public operator fun invoke(locked: Boolean = false): Mutex = Mutex(locked)
- }
-
- /**
- * Returns `true` when this mutex is locked.
- */
- public val isLocked: Boolean
-
- /**
- * Tries to lock this mutex, returning `false` if this mutex is already locked.
- *
- * @param owner Optional owner token for debugging. When `owner` is specified (non-null value) and this mutex
- * is already locked with the same token (same identity), this function throws [IllegalStateException].
- */
- public fun tryLock(owner: Any? = null): Boolean
-
- /**
- * Locks this mutex, suspending caller while the mutex is locked.
- *
- * 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.
- *
- * 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.
- *
- * This function can be used in [select] invocation with [onLock] clause.
- * Use [tryLock] to try acquire lock without waiting.
- *
- * @param owner Optional owner token for debugging. When `owner` is specified (non-null value) and this mutex
- * is already locked with the same token (same identity), this function throws [IllegalStateException].
- */
- public suspend fun lock(owner: Any? = null)
-
- /**
- * Clause for [select] expression of [lock] suspending function that selects when the mutex is locked.
- * Additional parameter for the clause in the `owner` (see [lock]) and when the clause is selected
- * the reference to this mutex is passed into the corresponding block.
- */
- public val onLock: SelectClause2<Any?, Mutex>
-
- /**
- * Checks mutex locked by owner
- *
- * @return `true` on mutex lock by owner, `false` if not locker or it is locked by different owner
- */
- public fun holdsLock(owner: Any): Boolean
-
- /**
- * Unlocks this mutex. Throws [IllegalStateException] if invoked on a mutex that is not locked.
- *
- * @param owner Optional owner token for debugging. When `owner` is specified (non-null value) and this mutex
- * was locked with the different token (by identity), this function throws [IllegalStateException].
- */
- public fun unlock(owner: Any? = null)
-}
-
-/**
- * Creates new [Mutex] instance.
- * The mutex created is fair: lock is granted in first come, first served order.
- *
- * @param locked initial state of the mutex.
- */
-public fun Mutex(locked: Boolean = false): Mutex = MutexImpl(locked)
-
-/**
- * Executes the given [action] under this mutex's lock.
- *
- * @param owner Optional owner token for debugging.
- *
- * @return the return value of the action.
- */
-public inline suspend fun <T> Mutex.withLock(owner: Any? = null, action: () -> T): T {
- lock(owner)
- try {
- return action()
- } finally {
- unlock(owner)
- }
-}
-
-/**
- * @suppress: **Deprecated**: binary compatibility with old code
- */
-@Deprecated("binary compatibility with old code", level = DeprecationLevel.HIDDEN)
-public suspend fun <T> Mutex.withLock(owner: Any? = null, action: suspend () -> T): T =
- withLock(owner) { action() }
-
-/**
- * @suppress: **Deprecated**: Use [withLock]
- */
-@Deprecated("Use `withLock(owner, action)", level = DeprecationLevel.HIDDEN)
-public suspend fun <T> Mutex.withLock(action: suspend () -> T): T =
- withLock { action() }
-
-/**
- * @suppress: **Deprecated**: Use [withLock]
- */
-@Deprecated("Use `withLock`", replaceWith = ReplaceWith("withLock(action)"))
-public suspend fun <T> Mutex.withMutex(action: suspend () -> T): T =
- withLock { action() }
-
-private val LOCK_FAIL = Symbol("LOCK_FAIL")
-private val ENQUEUE_FAIL = Symbol("ENQUEUE_FAIL")
-private val UNLOCK_FAIL = Symbol("UNLOCK_FAIL")
-private val SELECT_SUCCESS = Symbol("SELECT_SUCCESS")
-private val LOCKED = Symbol("LOCKED")
-private val UNLOCKED = Symbol("UNLOCKED")
-private val RESUME_QUIESCENT = Symbol("RESUME_QUIESCENT")
-private val RESUME_ACTIVE = Symbol("RESUME_ACTIVE")
-
-private val EmptyLocked = Empty(LOCKED)
-private val EmptyUnlocked = Empty(UNLOCKED)
-
-private class Empty(
- @JvmField val locked: Any
-) {
- override fun toString(): String = "Empty[$locked]"
-}
-
-internal class MutexImpl(locked: Boolean) : Mutex, SelectClause2<Any?, Mutex> {
- // State is: Empty | LockedQueue | OpDescriptor
- // shared objects while we have no waiters
- private val _state = atomic<Any?>(if (locked) EmptyLocked else EmptyUnlocked)
-
- // resumeNext is: RESUME_QUIESCENT | RESUME_ACTIVE | ResumeReq
- private val _resumeNext = atomic<Any>(RESUME_QUIESCENT)
-
- public override val isLocked: Boolean get() {
- _state.loop { state ->
- when (state) {
- is Empty -> return state.locked !== UNLOCKED
- is LockedQueue -> return true
- is OpDescriptor -> state.perform(this) // help
- else -> error("Illegal state $state")
- }
- }
- }
-
- // for tests ONLY
- internal val isLockedEmptyQueueState: Boolean get() {
- val state = _state.value
- return state is LockedQueue && state.isEmpty
- }
-
- public override fun tryLock(owner: Any?): Boolean {
- _state.loop { state ->
- when (state) {
- is Empty -> {
- if (state.locked !== UNLOCKED) return false
- val update = if (owner == null) EmptyLocked else Empty(owner)
- if (_state.compareAndSet(state, update)) return true
- }
- is LockedQueue -> {
- check(state.owner !== owner) { "Already locked by $owner" }
- return false
- }
- is OpDescriptor -> state.perform(this) // help
- else -> error("Illegal state $state")
- }
- }
- }
-
- public override suspend fun lock(owner: Any?) {
- // fast-path -- try lock
- if (tryLock(owner)) return
- // slow-path -- suspend
- return lockSuspend(owner)
- }
-
- private suspend fun lockSuspend(owner: Any?) = suspendAtomicCancellableCoroutine<Unit>(holdCancellability = true) sc@ { cont ->
- val waiter = LockCont(owner, cont)
- _state.loop { state ->
- when (state) {
- is Empty -> {
- if (state.locked !== UNLOCKED) { // try upgrade to queue & retry
- _state.compareAndSet(state, LockedQueue(state.locked))
- } else {
- // try lock
- val update = if (owner == null) EmptyLocked else Empty(owner)
- if (_state.compareAndSet(state, update)) { // locked
- cont.resume(Unit)
- return@sc
- }
- }
- }
- is LockedQueue -> {
- val curOwner = state.owner
- check(curOwner !== owner) { "Already locked by $owner" }
- if (state.addLastIf(waiter, { _state.value === state })) {
- // added to waiter list!
- cont.initCancellability() // make it properly cancellable
- cont.removeOnCancel(waiter)
- return@sc
- }
- }
- is OpDescriptor -> state.perform(this) // help
- else -> error("Illegal state $state")
- }
- }
- }
-
- override val onLock: SelectClause2<Any?, Mutex>
- get() = this
-
- // registerSelectLock
- @Suppress("PARAMETER_NAME_CHANGED_ON_OVERRIDE")
- override fun <R> registerSelectClause2(select: SelectInstance<R>, owner: Any?, block: suspend (Mutex) -> R) {
- while (true) { // lock-free loop on state
- if (select.isSelected) return
- val state = _state.value
- when (state) {
- is Empty -> {
- if (state.locked !== UNLOCKED) { // try upgrade to queue & retry
- _state.compareAndSet(state, LockedQueue(state.locked))
- } else {
- // try lock
- val failure = select.performAtomicTrySelect(TryLockDesc(this, owner))
- when {
- failure == null -> { // success
- block.startCoroutineUndispatched(receiver = this, completion = select.completion)
- return
- }
- failure === ALREADY_SELECTED -> return // already selected -- bail out
- failure === LOCK_FAIL -> {} // retry
- else -> error("performAtomicTrySelect(TryLockDesc) returned $failure")
- }
- }
- }
- is LockedQueue -> {
- check(state.owner !== owner) { "Already locked by $owner" }
- val enqueueOp = TryEnqueueLockDesc(this, owner, state, select, block)
- val failure = select.performAtomicIfNotSelected(enqueueOp)
- when {
- failure == null -> { // successfully enqueued
- select.disposeOnSelect(enqueueOp.node)
- return
- }
- failure === ALREADY_SELECTED -> return // already selected -- bail out
- failure === ENQUEUE_FAIL -> {} // retry
- else -> error("performAtomicIfNotSelected(TryEnqueueLockDesc) returned $failure")
- }
- }
- is OpDescriptor -> state.perform(this) // help
- else -> error("Illegal state $state")
- }
- }
- }
-
- private class TryLockDesc(
- @JvmField val mutex: MutexImpl,
- @JvmField val owner: Any?
- ) : AtomicDesc() {
- // This is Harris's RDCSS (Restricted Double-Compare Single Swap) operation
- private inner class PrepareOp(private val op: AtomicOp<*>) : OpDescriptor() {
- override fun perform(affected: Any?): Any? {
- val update: Any = if (op.isDecided) EmptyUnlocked else op // restore if was already decided
- (affected as MutexImpl)._state.compareAndSet(this, update)
- return null // ok
- }
- }
-
- override fun prepare(op: AtomicOp<*>): Any? {
- val prepare = PrepareOp(op)
- if (!mutex._state.compareAndSet(EmptyUnlocked, prepare)) return LOCK_FAIL
- return prepare.perform(mutex)
- }
-
- override fun complete(op: AtomicOp<*>, failure: Any?) {
- val update = if (failure != null) EmptyUnlocked else {
- if (owner == null) EmptyLocked else Empty(owner)
- }
- mutex._state.compareAndSet(op, update)
- }
- }
-
- private class TryEnqueueLockDesc<R>(
- @JvmField val mutex: MutexImpl,
- owner: Any?,
- queue: LockedQueue,
- select: SelectInstance<R>,
- block: suspend (Mutex) -> R
- ) : AddLastDesc<LockSelect<R>>(queue, LockSelect(owner, mutex, select, block)) {
- override fun onPrepare(affected: LockFreeLinkedListNode, next: LockFreeLinkedListNode): Any? {
- if (mutex._state.value !== queue) return ENQUEUE_FAIL
- return super.onPrepare(affected, next)
- }
- }
-
- public override fun holdsLock(owner: Any) =
- _state.value.let { state ->
- when (state) {
- is Empty -> state.locked === owner
- is LockedQueue -> state.owner === owner
- else -> false
- }
- }
-
- public override fun unlock(owner: Any?) {
- _state.loop { state ->
- when (state) {
- is Empty -> {
- if (owner == null)
- check(state.locked !== UNLOCKED) { "Mutex is not locked" }
- else
- check(state.locked === owner) { "Mutex is locked by ${state.locked} but expected $owner" }
- if (_state.compareAndSet(state, EmptyUnlocked)) return
- }
- is OpDescriptor -> state.perform(this)
- is LockedQueue -> {
- if (owner != null)
- check(state.owner === owner) { "Mutex is locked by ${state.owner} but expected $owner" }
- val waiter = state.removeFirstOrNull()
- if (waiter == null) {
- val op = UnlockOp(state)
- if (_state.compareAndSet(state, op) && op.perform(this) == null) return
- } else {
- val token = (waiter as LockWaiter).tryResumeLockWaiter()
- if (token != null) {
- // successfully resumed waiter that now is holding the lock
- // we must immediately transfer ownership to the next waiter, because this coroutine
- // might try to lock it again after unlock returns do to StackOverflow avoidance code
- // and its attempts to take a lock must be queued.
- state.owner = waiter.owner ?: LOCKED
- // StackOverflow avoidance code
- if (startResumeNext(waiter, token)) {
- waiter.completeResumeLockWaiter(token)
- finishResumeNext()
- }
- return
- }
- }
- }
- else -> error("Illegal state $state")
- }
- }
- }
-
- private class ResumeReq(
- @JvmField val waiter: LockWaiter,
- @JvmField val token: Any
- )
-
- private fun startResumeNext(waiter: LockWaiter, token: Any): Boolean {
- _resumeNext.loop { resumeNext ->
- when {
- resumeNext === RESUME_QUIESCENT -> {
- // this is never concurrent, because only one thread is holding mutex and trying to resume
- // next waiter, so no need to CAS here
- _resumeNext.value = RESUME_ACTIVE
- return true
- }
- resumeNext === RESUME_ACTIVE ->
- if (_resumeNext.compareAndSet(resumeNext, ResumeReq(waiter, token))) return false
- else -> error("Cannot happen")
- }
- }
- }
-
- private fun finishResumeNext() {
- // also a resumption loop to fulfill requests of inner resume invokes
- _resumeNext.loop { resumeNext ->
- when {
- resumeNext === RESUME_ACTIVE ->
- if (_resumeNext.compareAndSet(resumeNext, RESUME_QUIESCENT)) return
- resumeNext is ResumeReq -> {
- // this is never concurrently, only one thread is finishing, so no need to CAS here
- _resumeNext.value = RESUME_ACTIVE
- resumeNext.waiter.completeResumeLockWaiter(resumeNext.token)
- }
- else -> error("Cannot happen")
- }
- }
- }
-
- override fun toString(): String {
- _state.loop { state ->
- when (state) {
- is Empty -> return "Mutex[${state.locked}]"
- is OpDescriptor -> state.perform(this)
- is LockedQueue -> return "Mutex[${state.owner}]"
- else -> error("Illegal state $state")
- }
- }
- }
-
- private class LockedQueue(
- @JvmField var owner: Any
- ) : LockFreeLinkedListHead() {
- override fun toString(): String = "LockedQueue[$owner]"
- }
-
- private abstract class LockWaiter(
- @JvmField val owner: Any?
- ) : LockFreeLinkedListNode(), DisposableHandle {
- final override fun dispose() { remove() }
- abstract fun tryResumeLockWaiter(): Any?
- abstract fun completeResumeLockWaiter(token: Any)
- }
-
- private class LockCont(
- owner: Any?,
- @JvmField val cont: CancellableContinuation<Unit>
- ) : LockWaiter(owner) {
- override fun tryResumeLockWaiter() = cont.tryResume(Unit)
- override fun completeResumeLockWaiter(token: Any) = cont.completeResume(token)
- override fun toString(): String = "LockCont[$owner, $cont]"
- }
-
- private class LockSelect<R>(
- owner: Any?,
- @JvmField val mutex: Mutex,
- @JvmField val select: SelectInstance<R>,
- @JvmField val block: suspend (Mutex) -> R
- ) : LockWaiter(owner) {
- override fun tryResumeLockWaiter(): Any? = if (select.trySelect(null)) SELECT_SUCCESS else null
- override fun completeResumeLockWaiter(token: Any) {
- check(token === SELECT_SUCCESS)
- block.startCoroutine(receiver = mutex, completion = select.completion)
- }
- override fun toString(): String = "LockSelect[$owner, $mutex, $select]"
- }
-
- // atomic unlock operation that checks that waiters queue is empty
- private class UnlockOp(
- @JvmField val queue: LockedQueue
- ) : OpDescriptor() {
- override fun perform(affected: Any?): Any? {
- /*
- Note: queue cannot change while this UnlockOp is in progress, so all concurrent attempts to
- make a decision will reach it consistently. It does not matter what is a proposed
- decision when this UnlockOp is no longer active, because in this case the following CAS
- will fail anyway.
- */
- val success = queue.isEmpty
- val update: Any = if (success) EmptyUnlocked else queue
- (affected as MutexImpl)._state.compareAndSet(this@UnlockOp, update)
- /*
- `perform` invocation from the original `unlock` invocation may be coming too late, when
- some other thread had already helped to complete it (either successfully or not).
- That operation was unsuccessful if `state` was restored to this `queue` reference and
- that is what is being checked below.
- */
- return if (affected._state.value === queue) UNLOCK_FAIL else null
- }
- }
-}
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/JobTest.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/timeunit/TimeUnit.kt
similarity index 65%
copy from core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/JobTest.kt
copy to core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/timeunit/TimeUnit.kt
index 06f7c68..8c0a66c 100644
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/JobTest.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/timeunit/TimeUnit.kt
@@ -14,16 +14,10 @@
* limitations under the License.
*/
-package kotlinx.coroutines.experimental
+package kotlinx.coroutines.experimental.timeunit
-import kotlin.test.*
-
-class JobTest : TestBase() {
- @Test
- fun testMemoryRelease() {
- val job = Job()
- val n = 10_000_000 * stressTestMultiplier
- var fireCount = 0
- for (i in 0 until n) job.invokeOnCompletion { fireCount++ }.dispose()
- }
-}
\ No newline at end of file
+/**
+ * Time unit type alias for writing multiplatform code.
+ */
+@Suppress("ACTUAL_WITHOUT_EXPECT")
+public actual typealias TimeUnit = java.util.concurrent.TimeUnit
\ No newline at end of file
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/AsyncTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/AsyncJvmTest.kt
similarity index 98%
rename from core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/AsyncTest.kt
rename to core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/AsyncJvmTest.kt
index 7b93917..a2f4d5c 100644
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/AsyncTest.kt
+++ b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/AsyncJvmTest.kt
@@ -19,7 +19,7 @@
import kotlin.coroutines.experimental.*
import kotlin.test.*
-class AsyncTest : TestBase() {
+class AsyncJvmTest : TestBase() {
// This must be a common test but it fails on JS because of KT-21961
@Test
fun testAsyncWithFinally() = runTest {
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/AtomicCancellationTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/AtomicCancellationTest.kt
index 9d7a20a..1ea089a 100644
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/AtomicCancellationTest.kt
+++ b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/AtomicCancellationTest.kt
@@ -24,44 +24,6 @@
class AtomicCancellationTest : TestBase() {
@Test
- fun testLockAtomicCancel() = runBlocking {
- expect(1)
- val mutex = Mutex(true) // locked mutex
- val job = launch(coroutineContext, start = CoroutineStart.UNDISPATCHED) {
- expect(2)
- mutex.lock() // suspends
- expect(4) // should execute despite cancellation
- }
- expect(3)
- mutex.unlock() // unlock mutex first
- job.cancel() // cancel the job next
- yield() // now yield
- finish(5)
- }
-
- @Test
- fun testSelectLockAtomicCancel() = runBlocking {
- expect(1)
- val mutex = Mutex(true) // locked mutex
- val job = launch(coroutineContext, start = CoroutineStart.UNDISPATCHED) {
- expect(2)
- val result = select<String> { // suspends
- mutex.onLock {
- expect(4)
- "OK"
- }
- }
- assertEquals("OK", result)
- expect(5) // should execute despite cancellation
- }
- expect(3)
- mutex.unlock() // unlock mutex first
- job.cancel() // cancel the job next
- yield() // now yield
- finish(6)
- }
-
- @Test
fun testSendAtomicCancel() = runBlocking {
expect(1)
val channel = Channel<Int>()
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/CoroutinesTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/CoroutinesJvmTest.kt
similarity index 97%
rename from core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/CoroutinesTest.kt
rename to core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/CoroutinesJvmTest.kt
index 58e7bfe..b200015 100644
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/CoroutinesTest.kt
+++ b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/CoroutinesJvmTest.kt
@@ -19,7 +19,7 @@
import kotlin.coroutines.experimental.*
import kotlin.test.*
-class CoroutinesTest : TestBase() {
+class CoroutinesJvmTest : TestBase() {
@Test
fun testNotCancellableCodeWithExceptionCancelled() = runTest {
expect(1)
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/JobTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/JobStressTest.kt
similarity index 95%
rename from core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/JobTest.kt
rename to core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/JobStressTest.kt
index 06f7c68..50ef98e 100644
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/JobTest.kt
+++ b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/JobStressTest.kt
@@ -18,7 +18,7 @@
import kotlin.test.*
-class JobTest : TestBase() {
+class JobStressTest : TestBase() {
@Test
fun testMemoryRelease() {
val job = Job()
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/TestBase.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/TestBase.kt
index 4edc6e9..f6f087b 100644
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/TestBase.kt
+++ b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/TestBase.kt
@@ -14,9 +14,6 @@
* limitations under the License.
*/
-// :todo: Remove after transition to Kotlin 1.2.30+
-@file:Suppress("ACTUAL_FUNCTION_WITH_DEFAULT_ARGUMENTS")
-
package kotlinx.coroutines.experimental
import org.junit.After
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithContextTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithContextCommonPoolTest.kt
similarity index 95%
rename from core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithContextTest.kt
rename to core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithContextCommonPoolTest.kt
index ee26c07..993aeef 100644
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithContextTest.kt
+++ b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithContextCommonPoolTest.kt
@@ -18,7 +18,7 @@
import kotlin.test.*
-class WithContextTest : TestBase() {
+class WithContextCommonPoolTest : TestBase() {
@Test
fun testCommonPoolNoSuspend() = runTest {
expect(1)
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithTimeoutOrNullJvmTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithTimeoutOrNullJvmTest.kt
new file mode 100644
index 0000000..560d087
--- /dev/null
+++ b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithTimeoutOrNullJvmTest.kt
@@ -0,0 +1,36 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kotlinx.coroutines.experimental
+
+import kotlin.test.*
+
+class WithTimeoutOrNullJvmTest : TestBase() {
+ @Test
+ fun testOuterTimeoutFiredBeforeInner() = runTest {
+ val result = withTimeoutOrNull(100) {
+ Thread.sleep(200) // wait enough for outer timeout to fire
+ withContext(NonCancellable) { yield() } // give an event loop a chance to run and process that cancellation
+ withTimeoutOrNull(100) {
+ yield() // will cancel because of outer timeout
+ expectUnreached()
+ }
+ expectUnreached() // should not be reached, because it is outer timeout
+ }
+ // outer timeout results in null
+ assertEquals(null, result)
+ }
+}
\ No newline at end of file
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithTimeoutOrNullTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithTimeoutOrNullTest.kt
deleted file mode 100644
index 6c43f76..0000000
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithTimeoutOrNullTest.kt
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental
-
-import kotlin.test.*
-
-class WithTimeoutOrNullTest : TestBase() {
- @Test
- fun testNullOnTimeout() = runTest {
- expect(1)
- val result = withTimeoutOrNull(100) {
- expect(2)
- delay(1000)
- expectUnreached()
- "OK"
- }
- assertEquals(null, result)
- finish(3)
- }
-
- @Test
- fun testSuppressExceptionWithResult() = runTest {
- expect(1)
- val result = withTimeoutOrNull(100) {
- expect(2)
- try {
- delay(1000)
- } catch (e: CancellationException) {
- expect(3)
- }
- "OK"
- }
- assertEquals(null, result)
- finish(4)
- }
-
- @Test
- fun testSuppressExceptionWithAnotherException() = runTest(
- expected = { it is TestException }
- ) {
- expect(1)
- val result = withTimeoutOrNull(100) {
- expect(2)
- try {
- delay(1000)
- } catch (e: CancellationException) {
- finish(3)
- throw TestException()
- }
- expectUnreached()
- "OK"
- }
- expectUnreached()
- }
-
- @Test
- fun testOuterTimeoutFiredBeforeInner() = runTest {
- val result = withTimeoutOrNull(100) {
- Thread.sleep(200) // wait enough for outer timeout to fire
- withContext(NonCancellable) { yield() } // give an event loop a chance to run and process that cancellation
- withTimeoutOrNull(100) {
- yield() // will cancel because of outer timeout
- expectUnreached()
- }
- expectUnreached() // should not be reached, because it is outer timeout
- }
- // outer timeout results in null
- assertEquals(null, result)
- }
-
- private class TestException : Exception()
-}
\ No newline at end of file
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithTimeoutOrNullThreadDispatchTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithTimeoutOrNullThreadDispatchTest.kt
index df1387c..f5b8562 100644
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithTimeoutOrNullThreadDispatchTest.kt
+++ b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithTimeoutOrNullThreadDispatchTest.kt
@@ -47,7 +47,6 @@
}
}
-
@Test
fun testCancellationDispatchCustomNoDelay() {
// it also checks that there is at most once scheduled request in flight (no spurious concurrency)
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithTimeoutTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithTimeoutTest.kt
deleted file mode 100644
index a793a1e..0000000
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithTimeoutTest.kt
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-@file:Suppress("NAMED_ARGUMENTS_NOT_ALLOWED") // KT-21913
-
-package kotlinx.coroutines.experimental
-
-import kotlin.test.*
-import java.io.IOException
-
-class WithTimeoutTest : TestBase() {
- @Test
- fun testExceptionOnTimeout() = runTest {
- expect(1)
- try {
- withTimeout(100) {
- expect(2)
- delay(1000)
- expectUnreached()
- "OK"
- }
- } catch (e: CancellationException) {
- assertEquals("Timed out waiting for 100 MILLISECONDS", e.message)
- finish(3)
- }
- }
-
- @Test
- fun testSuppressExceptionWithResult() = runTest(
- expected = { it is CancellationException }
- ) {
- expect(1)
- val result = withTimeout(100) {
- expect(2)
- try {
- delay(1000)
- } catch (e: CancellationException) {
- finish(3)
- }
- "OK"
- }
- expectUnreached()
- }
-
- @Test
- fun testSuppressExceptionWithAnotherException() = runTest(
- expected = { it is IOException }
- ) {
- expect(1)
- withTimeout(100) {
- expect(2)
- try {
- delay(1000)
- } catch (e: CancellationException) {
- finish(3)
- throw IOException(e)
- }
- expectUnreached()
- "OK"
- }
- expectUnreached()
- }
-}
\ No newline at end of file
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/BroadcastChannelMultiReceiveStressTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/BroadcastChannelMultiReceiveStressTest.kt
index 7f341c1..60bdeb1 100644
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/BroadcastChannelMultiReceiveStressTest.kt
+++ b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/BroadcastChannelMultiReceiveStressTest.kt
@@ -18,10 +18,10 @@
import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.selects.*
+import kotlinx.coroutines.experimental.timeunit.*
import org.junit.*
import org.junit.runner.*
import org.junit.runners.*
-import java.util.concurrent.*
import java.util.concurrent.atomic.*
import kotlin.coroutines.experimental.*
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/BroadcastChannelSubStressTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/BroadcastChannelSubStressTest.kt
index 21b238c..cfef040 100644
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/BroadcastChannelSubStressTest.kt
+++ b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/BroadcastChannelSubStressTest.kt
@@ -17,10 +17,10 @@
package kotlinx.coroutines.experimental.channels
import kotlinx.coroutines.experimental.*
+import kotlinx.coroutines.experimental.timeunit.*
import org.junit.*
import org.junit.runner.*
import org.junit.runners.*
-import java.util.concurrent.*
import java.util.concurrent.atomic.*
import kotlin.coroutines.experimental.*
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectBuilderImplTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectBuilderImplTest.kt
deleted file mode 100644
index 39e289d..0000000
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectBuilderImplTest.kt
+++ /dev/null
@@ -1,136 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental.selects
-
-import org.junit.Test
-import kotlin.coroutines.experimental.Continuation
-import kotlin.coroutines.experimental.CoroutineContext
-import kotlin.coroutines.experimental.EmptyCoroutineContext
-import kotlin.coroutines.experimental.intrinsics.COROUTINE_SUSPENDED
-
-class SelectBuilderImplTest {
- @Test
- fun testIdempotentSelectResumeInline() {
- var resumed = false
- val delegate = object : Continuation<String> {
- override val context: CoroutineContext get() = EmptyCoroutineContext
- override fun resume(value: String) {
- check(value === "OK")
- resumed = true
- }
- override fun resumeWithException(exception: Throwable) { error("Should not happen") }
- }
- val c = SelectBuilderImpl<String>(delegate)
- // still running builder
- check(!c.isSelected)
- check(c.trySelect("SELECT"))
- check(c.isSelected)
- check(!c.trySelect("OTHER"))
- check(c.trySelect("SELECT"))
- c.completion.resume("OK")
- check(!resumed) // still running builder, didn't invoke delegate
- check(c.isSelected)
- check(!c.trySelect("OTHER"))
- check(c.trySelect("SELECT"))
- check(c.getResult() === "OK") // then builder returns
- }
-
- @Test
- fun testIdempotentSelectResumeSuspended() {
- var resumed = false
- val delegate = object : Continuation<String> {
- override val context: CoroutineContext get() = EmptyCoroutineContext
- override fun resume(value: String) {
- check(value === "OK")
- resumed = true
- }
- override fun resumeWithException(exception: Throwable) { error("Should not happen") }
- }
- val c = SelectBuilderImpl<String>(delegate)
- check(c.getResult() === COROUTINE_SUSPENDED) // suspend first
- check(!c.isSelected)
- check(c.trySelect("SELECT"))
- check(c.isSelected)
- check(!c.trySelect("OTHER"))
- check(c.trySelect("SELECT"))
- check(!resumed)
- c.completion.resume("OK")
- check(resumed)
- check(c.isSelected)
- check(!c.trySelect("OTHER"))
- check(c.trySelect("SELECT"))
- }
-
- @Test
- fun testIdempotentSelectResumeWithExceptionInline() {
- var resumed = false
- val delegate = object : Continuation<String> {
- override val context: CoroutineContext get() = EmptyCoroutineContext
- override fun resume(value: String) { error("Should not happen") }
- override fun resumeWithException(exception: Throwable) {
- check(exception is TestException)
- resumed = true
- }
- }
- val c = SelectBuilderImpl<String>(delegate)
- // still running builder
- check(!c.isSelected)
- check(c.trySelect("SELECT"))
- check(c.isSelected)
- check(!c.trySelect("OTHER"))
- check(c.trySelect("SELECT"))
- c.completion.resumeWithException(TestException())
- check(!resumed) // still running builder, didn't invoke delegate
- check(c.isSelected)
- check(!c.trySelect("OTHER"))
- check(c.trySelect("SELECT"))
- try {
- c.getResult() // the builder should throw exception
- error("Failed")
- } catch (e: Throwable) {
- check(e is TestException)
- }
- }
-
- @Test
- fun testIdempotentSelectResumeWithExceptionSuspended() {
- var resumed = false
- val delegate = object : Continuation<String> {
- override val context: CoroutineContext get() = EmptyCoroutineContext
- override fun resume(value: String) { error("Should not happen") }
- override fun resumeWithException(exception: Throwable) {
- check(exception is TestException)
- resumed = true
- }
- }
- val c = SelectBuilderImpl<String>(delegate)
- check(c.getResult() === COROUTINE_SUSPENDED) // suspend first
- check(!c.isSelected)
- check(c.trySelect("SELECT"))
- check(c.isSelected)
- check(!c.trySelect("OTHER"))
- check(c.trySelect("SELECT"))
- check(!resumed)
- c.completion.resumeWithException(TestException())
- check(resumed)
- check(c.isSelected)
- check(!c.trySelect("OTHER"))
- check(c.trySelect("SELECT"))
- }
-
- class TestException : Throwable()
-}
\ No newline at end of file
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectDeferredTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectDeferredTest.kt
deleted file mode 100644
index 3ab8955..0000000
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectDeferredTest.kt
+++ /dev/null
@@ -1,151 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental.selects
-
-import kotlinx.coroutines.experimental.*
-import org.junit.*
-import org.junit.Assert.*
-import kotlin.coroutines.experimental.*
-
-class SelectDeferredTest : TestBase() {
- @Test
- fun testSimpleReturnsImmediately() = runBlocking<Unit> {
- expect(1)
- val d1 = async<Int>(coroutineContext) {
- expect(3)
- 42
- }
- expect(2)
- val res = select<String> {
- d1.onAwait { v ->
- expect(4)
- assertEquals(42, v)
- "OK"
- }
- }
- expect(5)
- assertEquals("OK", res)
- finish(6)
- }
-
- @Test
- fun testSimpleWithYield() = runBlocking<Unit> {
- expect(1)
- val d1 = async<Int>(coroutineContext) {
- expect(3)
- 42
- }
- launch(coroutineContext) {
- expect(4)
- yield() // back to main
- expect(6)
- }
- expect(2)
- val res = select<String> {
- d1.onAwait { v ->
- expect(5)
- assertEquals(42, v)
- yield() // to launch
- expect(7)
- "OK"
- }
- }
- finish(8)
- assertEquals("OK", res)
- }
-
- @Test
- fun testSelectIncompleteLazy() = runBlocking<Unit> {
- expect(1)
- val d1 = async(coroutineContext, CoroutineStart.LAZY) {
- expect(5)
- 42
- }
- launch(coroutineContext) {
- expect(3)
- val res = select<String> {
- d1.onAwait { v ->
- expect(7)
- assertEquals(42, v)
- "OK"
- }
- }
- expect(8)
- assertEquals("OK", res)
- }
- expect(2)
- yield() // to launch
- expect(4)
- yield() // to started async
- expect(6)
- yield() // to triggered select
- finish(9)
- }
-
- @Test
- fun testSelectTwo() = runBlocking<Unit> {
- expect(1)
- val d1 = async<String>(coroutineContext) {
- expect(3)
- yield() // to the other deffered
- expect(5)
- yield() // to fired select
- expect(7)
- "d1"
- }
- val d2 = async<String>(coroutineContext) {
- expect(4)
- "d2" // returns result
- }
- expect(2)
- val res = select<String> {
- d1.onAwait {
- expectUnreached()
- "FAIL"
- }
- d2.onAwait { v2 ->
- expect(6)
- assertEquals("d2", v2)
- yield() // to first deferred
- expect(8)
- "OK"
- }
- }
- assertEquals("OK", res)
- finish(9)
- }
-
- @Test
- fun testSelectCancel() = runTest(
- expected = { it is JobCancellationException }
- ) {
- expect(1)
- val d = CompletableDeferred<String>()
- launch (coroutineContext) {
- finish(3)
- d.cancel() // will cancel after select starts
- }
- expect(2)
- select<Unit> {
- d.onAwait {
- expectUnreached() // will not select
- }
- }
- expectUnreached()
- }
-
-}
\ No newline at end of file
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectJobTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectJobTest.kt
deleted file mode 100644
index 16cffda..0000000
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectJobTest.kt
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental.selects
-
-import kotlinx.coroutines.experimental.*
-import org.junit.*
-import org.junit.Assert.*
-import kotlin.coroutines.experimental.*
-
-class SelectJobTest : TestBase() {
- @Test
- fun testSelectCompleted() = runTest {
- expect(1)
- launch(coroutineContext) { // makes sure we don't yield to it earlier
- finish(4) // after main exits
- }
- val job = Job()
- job.cancel()
- select<Unit> {
- job.onJoin {
- expect(2)
- }
- }
- expect(3)
- // will wait for the first coroutine
- }
-
- @Test
- fun testSelectIncomplete() = runTest {
- expect(1)
- val job = Job()
- launch(coroutineContext) { // makes sure we don't yield to it earlier
- expect(3)
- val res = select<String> {
- job.onJoin {
- expect(6)
- "OK"
- }
- }
- expect(7)
- assertEquals("OK", res)
- }
- expect(2)
- yield()
- expect(4)
- job.cancel()
- expect(5)
- yield()
- finish(8)
- }
-
- @Test
- fun testSelectLazy() = runTest {
- expect(1)
- val job = launch(coroutineContext, CoroutineStart.LAZY) {
- expect(2)
- }
- val res = select<String> {
- job.onJoin {
- expect(3)
- "OK"
- }
- }
- finish(4)
- assertEquals("OK", res)
- }
-}
\ No newline at end of file
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectMutexStressTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectMutexStressTest.kt
new file mode 100644
index 0000000..d7be9ee
--- /dev/null
+++ b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectMutexStressTest.kt
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kotlinx.coroutines.experimental.selects
+
+import kotlinx.coroutines.experimental.*
+import kotlinx.coroutines.experimental.sync.*
+import kotlin.test.*
+
+class SelectMutexStressTest : TestBase() {
+ @Test
+ fun testSelectCancelledResourceRelease() = runTest {
+ val n = 1_000 * stressTestMultiplier
+ val mutex = Mutex(true) as MutexImpl // locked
+ expect(1)
+ repeat(n) { i ->
+ val job = launch(kotlin.coroutines.experimental.coroutineContext) {
+ expect(i + 2)
+ select<Unit> {
+ mutex.onLock {
+ expectUnreached() // never able to lock
+ }
+ }
+ }
+ yield() // to the launched job, so that it suspends
+ job.cancel() // cancel the job and select
+ yield() // so it can cleanup after itself
+ }
+ assertTrue(mutex.isLocked)
+ assertTrue(mutex.isLockedEmptyQueueState)
+ finish(n + 2)
+ }
+}
\ No newline at end of file
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectMutexTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectMutexTest.kt
deleted file mode 100644
index 96c94a4..0000000
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectMutexTest.kt
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental.selects
-
-import kotlinx.coroutines.experimental.*
-import kotlinx.coroutines.experimental.sync.*
-import org.junit.*
-import org.junit.Assert.*
-import kotlin.coroutines.experimental.*
-
-class SelectMutexTest : TestBase() {
- @Test
- fun testSelectLock() = runTest {
- val mutex = Mutex()
- expect(1)
- launch(coroutineContext) { // ensure that it is not scheduled earlier than needed
- finish(4) // after main exits
- }
- val res = select<String> {
- mutex.onLock {
- assertTrue(mutex.isLocked)
- expect(2)
- "OK"
- }
- }
- assertEquals("OK", res)
- expect(3)
- // will wait for the first coroutine
- }
-
- @Test
- fun testSelectLockWait() = runTest {
- val mutex = Mutex(true) // locked
- expect(1)
- launch(coroutineContext) {
- expect(3)
- val res = select<String> { // will suspended
- mutex.onLock {
- assertTrue(mutex.isLocked)
- expect(6)
- "OK"
- }
- }
- assertEquals("OK", res)
- expect(7)
- }
- expect(2)
- yield() // to launched coroutine
- expect(4)
- mutex.unlock()
- expect(5)
- yield() // to resumed select
- finish(8)
- }
-
- @Test
- fun testSelectCancelledResourceRelease() = runTest {
- val n = 1_000 * stressTestMultiplier
- val mutex = Mutex(true) as MutexImpl // locked
- expect(1)
- repeat(n) { i ->
- val job = launch(coroutineContext) {
- expect(i + 2)
- select<Unit> {
- mutex.onLock {
- expectUnreached() // never able to lock
- }
- }
- }
- yield() // to the launched job, so that it suspends
- job.cancel() // cancel the job and select
- yield() // so it can cleanup after itself
- }
- assertTrue(mutex.isLocked)
- assertTrue(mutex.isLockedEmptyQueueState)
- finish(n + 2)
- }
-}
\ No newline at end of file
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectTimeoutTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectTimeoutTest.kt
deleted file mode 100644
index b8842c1..0000000
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectTimeoutTest.kt
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental.selects
-
-import kotlinx.coroutines.experimental.TestBase
-import kotlinx.coroutines.experimental.runBlocking
-import org.hamcrest.MatcherAssert.assertThat
-import org.hamcrest.core.IsEqual
-import org.junit.Test
-
-class SelectTimeoutTest : TestBase() {
- @Test
- fun testBasic() = runBlocking {
- expect(1)
- val result = select<String> {
- onTimeout(1000) {
- expectUnreached()
- "FAIL"
- }
- onTimeout(100) {
- expect(2)
- "OK"
- }
- onTimeout(500) {
- expectUnreached()
- "FAIL"
- }
- }
- assertThat(result, IsEqual("OK"))
- finish(3)
- }
-}
\ No newline at end of file
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/sync/MutexStressTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/sync/MutexStressTest.kt
new file mode 100644
index 0000000..720bd3b
--- /dev/null
+++ b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/sync/MutexStressTest.kt
@@ -0,0 +1,42 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kotlinx.coroutines.experimental.sync
+
+import kotlinx.coroutines.experimental.*
+import kotlin.test.*
+
+class MutexStressTest : TestBase() {
+ @Test
+ fun testStress() = runTest {
+ val n = 1000 * stressTestMultiplier
+ val k = 100
+ var shared = 0
+ val mutex = Mutex()
+ val jobs = List(n) {
+ launch(CommonPool) {
+ repeat(k) {
+ mutex.lock()
+ shared++
+ mutex.unlock()
+ }
+ }
+ }
+ jobs.forEach { it.join() }
+ println("Shared value = $shared")
+ assertEquals(n * k, shared)
+ }
+}
\ No newline at end of file
diff --git a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/sync/MutexTest.kt b/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/sync/MutexTest.kt
deleted file mode 100644
index 1d3fd30..0000000
--- a/core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/sync/MutexTest.kt
+++ /dev/null
@@ -1,144 +0,0 @@
-/*
- * Copyright 2016-2017 JetBrains s.r.o.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kotlinx.coroutines.experimental.sync
-
-import kotlinx.coroutines.experimental.*
-import org.hamcrest.core.*
-import org.junit.*
-import org.junit.Assert.*
-import kotlin.coroutines.experimental.*
-
-class MutexTest : TestBase() {
- @Test
- fun testSimple() = runBlocking<Unit> {
- val mutex = Mutex()
- expect(1)
- launch(coroutineContext) {
- expect(4)
- mutex.lock() // suspends
- expect(7) // now got lock
- mutex.unlock()
- expect(8)
- }
- expect(2)
- mutex.lock() // locked
- expect(3)
- yield() // yield to child
- expect(5)
- mutex.unlock()
- expect(6)
- yield() // now child has lock
- finish(9)
- }
-
- @Test
- fun tryLockTest() {
- val mutex = Mutex()
- assertFalse(mutex.isLocked)
- assertTrue(mutex.tryLock())
- assertTrue(mutex.isLocked)
- assertFalse(mutex.tryLock())
- assertTrue(mutex.isLocked)
- mutex.unlock()
- assertFalse(mutex.isLocked)
- assertTrue(mutex.tryLock())
- assertTrue(mutex.isLocked)
- assertFalse(mutex.tryLock())
- assertTrue(mutex.isLocked)
- mutex.unlock()
- assertFalse(mutex.isLocked)
- }
-
- @Test
- fun withLockTest() = runBlocking {
- val mutex = Mutex()
- assertFalse(mutex.isLocked)
- mutex.withLock {
- assertTrue(mutex.isLocked)
- }
- assertFalse(mutex.isLocked)
- }
-
- @Test
- fun testStress() = runBlocking<Unit> {
- val n = 1000 * stressTestMultiplier
- val k = 100
- var shared = 0
- val mutex = Mutex()
- val jobs = List(n) {
- launch(CommonPool) {
- repeat(k) {
- mutex.lock()
- shared++
- mutex.unlock()
- }
- }
- }
- jobs.forEach { it.join() }
- println("Shared value = $shared")
- assertEquals(n * k, shared)
- }
-
- @Test
- fun testUnconfinedStackOverflow() {
- val waiters = 10000
- val mutex = Mutex(true)
- var done = 0
- repeat(waiters) {
- launch(Unconfined) { // a lot of unconfined waiters
- mutex.withLock {
- done++
- }
- }
- }
- mutex.unlock() // should not produce StackOverflowError
- assertThat(done, IsEqual(waiters))
- }
-
- @Test
- fun holdLock() = runBlocking {
- val mutex = Mutex()
- val firstOwner = Any()
- val secondOwner = Any()
-
- // no lock
- assertFalse(mutex.holdsLock(firstOwner))
- assertFalse(mutex.holdsLock(secondOwner))
-
- // owner firstOwner
- mutex.lock(firstOwner)
- val secondLockJob = launch(CommonPool) {
- mutex.lock(secondOwner)
- }
-
- assertTrue(mutex.holdsLock(firstOwner))
- assertFalse(mutex.holdsLock(secondOwner))
-
- // owner secondOwner
- mutex.unlock(firstOwner)
- secondLockJob.join()
-
- assertFalse(mutex.holdsLock(firstOwner))
- assertTrue(mutex.holdsLock(secondOwner))
-
- mutex.unlock(secondOwner)
-
- // no lock
- assertFalse(mutex.holdsLock(firstOwner))
- assertFalse(mutex.holdsLock(secondOwner))
- }
-}
\ No newline at end of file