-
Notifications
You must be signed in to change notification settings - Fork 1.9k
/
Copy pathCancellableContinuation.kt
270 lines (247 loc) · 11.8 KB
/
CancellableContinuation.kt
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
/*
* Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
*/
package kotlinx.coroutines
import kotlinx.coroutines.internal.*
import kotlin.coroutines.*
import kotlin.coroutines.intrinsics.*
// --------------- cancellable continuations ---------------
/**
* Cancellable continuation. It 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].
* Invocation of [resume] in _cancelled_ state is ignored (it is a trivial race between resume from the continuation owner and
* outer job cancellation and cancellation wins).
* Invocation of [resumeWithException] in _cancelled_ state triggers exception handling of passed exception.
*
* ```
* +-----------+ resume +---------+
* | Active | ----------> | Resumed |
* +-----------+ +---------+
* |
* | cancel
* V
* +-----------+
* | Cancelled |
* +-----------+
*
* ```
*/
public interface CancellableContinuation<in T> : Continuation<T> {
/**
* Returns `true` when this continuation is active -- it has not completed or cancelled yet.
*/
public val isActive: Boolean
/**
* Returns `true` when this continuation has completed for any reason. A continuation
* that was cancelled is also considered complete.
*/
public val isCompleted: Boolean
/**
* Returns `true` if this continuation was [cancelled][cancel].
*
* It implies that [isActive] is `false` and [isCompleted] is `true`.
*/
public 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.**
*/
@InternalCoroutinesApi
public 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.**
*/
@InternalCoroutinesApi
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.**
*/
@InternalCoroutinesApi
public fun completeResume(token: Any)
/**
* Legacy function that turned on cancellation behavior in [suspendCancellableCoroutine] before kotlinx.coroutines 1.1.0.
* This function does nothing and is left only for binary compatibility with old compiled code.
*
* @suppress **Deprecated**: This function is no longer used.
* It is left for binary compatibility with code compiled before kotlinx.coroutines 1.1.0.
*/
@Deprecated(
level = DeprecationLevel.HIDDEN,
message = "This function is no longer used. " +
"It is left for binary compatibility with code compiled before kotlinx.coroutines 1.1.0. "
)
@InternalCoroutinesApi
public 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.
*/
public fun cancel(cause: Throwable? = null): Boolean
/**
* Registers handler that is **synchronously** invoked once on cancellation (both regular and exceptional) of this continuation.
* When the continuation is already cancelled, then the handler is immediately invoked
* with cancellation exception. Otherwise, the handler will be invoked once on cancellation if this
* continuation is cancelled.
*
* Installed [handler] should not throw any exceptions.
* If it does, they will get caught, wrapped into [CompletionHandlerException] and
* processed as uncaught exception in the context of the current coroutine
* (see [CoroutineExceptionHandler]).
*
* At most one [handler] can be installed on one continuation.
*
* **Note**: Implementation of `CompletionHandler` must be fast, non-blocking, and thread-safe.
* This handler can be invoked concurrently with the surrounding code.
* There is no guarantee on the execution context in which the [handler] is invoked.
*/
public fun invokeOnCancellation(handler: CompletionHandler)
/**
* 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**.
*
* **Note: This function is experimental.** Its signature general code may be changed in the future.
*/
@ExperimentalCoroutinesApi
public 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**.
*
* **Note: This function is experimental.** Its signature general code may be changed in the future.
*/
@ExperimentalCoroutinesApi
public fun CoroutineDispatcher.resumeUndispatchedWithException(exception: Throwable)
/**
* Resumes this continuation with a given [value] and calls the specified [onCancellation]
* handler when resumed too late (when continuation was already cancelled) or when resumed
* successfully (before cancellation), but coroutine's job was cancelled before it had a
* chance to run in its dispatcher, so that suspended function threw an exception
* instead of returning this value.
*
* Installed [onCancellation] handler should not throw any exceptions.
* If it does, they will get caught, wrapped into [CompletionHandlerException] and
* processed as uncaught exception in the context of the current coroutine
* (see [CoroutineExceptionHandler]).
*
* This function shall be used when resuming with a resource that must be closed by the
* code that had called the corresponding suspending function, e.g.:
*
* ```
* continuation.resume(resource) {
* resource.close()
* }
* ```
*
* **Note**: Implementation of [onCancellation] handler must be fast, non-blocking, and thread-safe.
* This handler can be invoked concurrently with the surrounding code.
* There is no guarantee on the execution context in which the [onCancellation] handler is invoked.
*/
@ExperimentalCoroutinesApi // since 1.2.0, tentatively graduates in 1.3.0
public fun resume(value: T, onCancellation: (cause: Throwable) -> Unit)
}
/**
* 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.
*/
public suspend inline fun <T> suspendCancellableCoroutine(
crossinline block: (CancellableContinuation<T>) -> Unit
): T =
suspendCoroutineUninterceptedOrReturn { uCont ->
val cancellable = CancellableContinuationImpl(uCont.intercepted(), resumeMode = MODE_CANCELLABLE)
// NOTE: Before version 1.1.0 the following invocation was inlined here, so invocation of this
// method indicates that the code was compiled by kotlinx.coroutines < 1.1.0
// 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.
*
* @suppress **This an internal API and should not be used from general code.**
*/
@InternalCoroutinesApi
public suspend inline fun <T> suspendAtomicCancellableCoroutine(
crossinline block: (CancellableContinuation<T>) -> Unit
): T =
suspendCoroutineUninterceptedOrReturn { uCont ->
val cancellable = CancellableContinuationImpl(uCont.intercepted(), resumeMode = MODE_ATOMIC_DEFAULT)
block(cancellable)
cancellable.getResult()
}
/**
* @suppress **Deprecated**
*/
@Deprecated(
message = "holdCancellability parameter is deprecated and is no longer used",
replaceWith = ReplaceWith("suspendAtomicCancellableCoroutine(block)")
)
@InternalCoroutinesApi
public suspend inline fun <T> suspendAtomicCancellableCoroutine(
holdCancellability: Boolean = false,
crossinline block: (CancellableContinuation<T>) -> Unit
): T =
suspendAtomicCancellableCoroutine(block)
/**
* Removes a given node on cancellation.
*/
internal fun CancellableContinuation<*>.removeOnCancellation(node: LockFreeLinkedListNode) =
invokeOnCancellation(handler = RemoveOnCancel(node).asHandler)
/**
* Disposes a specified [handle] when this continuation is cancelled.
*
* This is a shortcut for the following code with slightly more efficient implementation (one fewer object created).
* ```
* invokeOnCancellation { handle.dispose() }
* ```
*
* @suppress **This an internal API and should not be used from general code.**
*/
@InternalCoroutinesApi
public fun CancellableContinuation<*>.disposeOnCancellation(handle: DisposableHandle) =
invokeOnCancellation(handler = DisposeOnCancel(handle).asHandler)
// --------------- implementation details ---------------
private class RemoveOnCancel(private val node: LockFreeLinkedListNode) : CancelHandler() {
override fun invoke(cause: Throwable?) { node.remove() }
override fun toString() = "RemoveOnCancel[$node]"
}
private class DisposeOnCancel(private val handle: DisposableHandle) : CancelHandler() {
override fun invoke(cause: Throwable?) = handle.dispose()
override fun toString(): String = "DisposeOnCancel[$handle]"
}