blob: 409602fba3c78c11a27e0a7f64dd6af13255e82e [file] [log] [blame] [view]
Roman Elizarov43e90112017-05-10 11:25:20 +03001<!--- INCLUDE .*/example-([a-z]+)-([0-9a-z]+)\.kt
Roman Elizarova5e653f2017-02-13 13:49:55 +03002/*
3 * Copyright 2016-2017 JetBrains s.r.o.
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 */
Roman Elizarovf16fd272017-02-07 11:26:00 +030017
Roman Elizarova5e653f2017-02-13 13:49:55 +030018// This file was automatically generated from coroutines-guide.md by Knit tool. Do not edit.
19package guide.$$1.example$$2
Roman Elizarovf16fd272017-02-07 11:26:00 +030020
Roman Elizarova5e653f2017-02-13 13:49:55 +030021import kotlinx.coroutines.experimental.*
Roman Elizarovf16fd272017-02-07 11:26:00 +030022-->
Roman Elizarove8d79342017-08-29 15:21:21 +030023<!--- KNIT core/kotlinx-coroutines-core/src/test/kotlin/guide/.*\.kt -->
24<!--- TEST_OUT core/kotlinx-coroutines-core/src/test/kotlin/guide/test/GuideTest.kt
Roman Elizarov731f0ad2017-02-22 20:48:45 +030025// This file was automatically generated from coroutines-guide.md by Knit tool. Do not edit.
26package guide.test
27
28import org.junit.Test
29
30class GuideTest {
31-->
Roman Elizarovf16fd272017-02-07 11:26:00 +030032
Roman Elizarov7deefb82017-01-31 10:33:17 +030033# Guide to kotlinx.coroutines by example
34
35This is a short guide on core features of `kotlinx.coroutines` with a series of examples.
36
Roman Elizarov2a638922017-03-04 10:22:43 +030037## Introduction and setup
38
39Kotlin, as a language, provides only minimal low-level APIs in its standard library to enable various other
40libraries to utilize coroutines. Unlike many other languages with similar capabilities, `async` and `await`
41are not keywords in Kotlin and are not even part of its standard library.
42
Robert Hencke497d3432017-04-11 00:14:29 -040043`kotlinx.coroutines` is one such rich library. It contains a number of high-level
Roman Elizarov2a638922017-03-04 10:22:43 +030044coroutine-enabled primitives that this guide covers, including `async` and `await`.
45You need to add a dependency on `kotlinx-coroutines-core` module as explained
46[here](README.md#using-in-your-projects) to use primitives from this guide in your projects.
47
Roman Elizarov1293ccd2017-02-01 18:49:54 +030048## Table of contents
49
Roman Elizarovfa7723e2017-02-06 11:17:51 +030050<!--- TOC -->
51
Roman Elizarov1293ccd2017-02-01 18:49:54 +030052* [Coroutine basics](#coroutine-basics)
53 * [Your first coroutine](#your-first-coroutine)
54 * [Bridging blocking and non-blocking worlds](#bridging-blocking-and-non-blocking-worlds)
55 * [Waiting for a job](#waiting-for-a-job)
56 * [Extract function refactoring](#extract-function-refactoring)
57 * [Coroutines ARE light-weight](#coroutines-are-light-weight)
58 * [Coroutines are like daemon threads](#coroutines-are-like-daemon-threads)
59* [Cancellation and timeouts](#cancellation-and-timeouts)
60 * [Cancelling coroutine execution](#cancelling-coroutine-execution)
61 * [Cancellation is cooperative](#cancellation-is-cooperative)
62 * [Making computation code cancellable](#making-computation-code-cancellable)
63 * [Closing resources with finally](#closing-resources-with-finally)
64 * [Run non-cancellable block](#run-non-cancellable-block)
65 * [Timeout](#timeout)
66* [Composing suspending functions](#composing-suspending-functions)
67 * [Sequential by default](#sequential-by-default)
Roman Elizarov32d95322017-02-09 15:57:31 +030068 * [Concurrent using async](#concurrent-using-async)
69 * [Lazily started async](#lazily-started-async)
70 * [Async-style functions](#async-style-functions)
Roman Elizarov2f6d7c92017-02-03 15:16:07 +030071* [Coroutine context and dispatchers](#coroutine-context-and-dispatchers)
Roman Elizarovfa7723e2017-02-06 11:17:51 +030072 * [Dispatchers and threads](#dispatchers-and-threads)
Roman Elizarov2f6d7c92017-02-03 15:16:07 +030073 * [Unconfined vs confined dispatcher](#unconfined-vs-confined-dispatcher)
74 * [Debugging coroutines and threads](#debugging-coroutines-and-threads)
75 * [Jumping between threads](#jumping-between-threads)
76 * [Job in the context](#job-in-the-context)
77 * [Children of a coroutine](#children-of-a-coroutine)
78 * [Combining contexts](#combining-contexts)
Roman Elizarov8b38fa22017-09-27 17:44:31 +030079 * [Parental responsibilities](#parental-responsibilities)
Roman Elizarov2f6d7c92017-02-03 15:16:07 +030080 * [Naming coroutines for debugging](#naming-coroutines-for-debugging)
Roman Elizarov2fd7cb32017-02-11 23:18:59 +030081 * [Cancellation via explicit job](#cancellation-via-explicit-job)
Roman Elizarovb7721cf2017-02-03 19:23:08 +030082* [Channels](#channels)
83 * [Channel basics](#channel-basics)
84 * [Closing and iteration over channels](#closing-and-iteration-over-channels)
85 * [Building channel producers](#building-channel-producers)
86 * [Pipelines](#pipelines)
87 * [Prime numbers with pipeline](#prime-numbers-with-pipeline)
88 * [Fan-out](#fan-out)
89 * [Fan-in](#fan-in)
90 * [Buffered channels](#buffered-channels)
Roman Elizarovb0517ba2017-02-27 14:03:14 +030091 * [Channels are fair](#channels-are-fair)
Roman Elizarovf5bc0472017-02-22 11:38:13 +030092* [Shared mutable state and concurrency](#shared-mutable-state-and-concurrency)
93 * [The problem](#the-problem)
Roman Elizarov1e459602017-02-27 11:05:17 +030094 * [Volatiles are of no help](#volatiles-are-of-no-help)
Roman Elizarovf5bc0472017-02-22 11:38:13 +030095 * [Thread-safe data structures](#thread-safe-data-structures)
Roman Elizarov1e459602017-02-27 11:05:17 +030096 * [Thread confinement fine-grained](#thread-confinement-fine-grained)
97 * [Thread confinement coarse-grained](#thread-confinement-coarse-grained)
Roman Elizarovf5bc0472017-02-22 11:38:13 +030098 * [Mutual exclusion](#mutual-exclusion)
99 * [Actors](#actors)
Roman Elizarovd4dcbe22017-02-22 09:57:46 +0300100* [Select expression](#select-expression)
101 * [Selecting from channels](#selecting-from-channels)
102 * [Selecting on close](#selecting-on-close)
103 * [Selecting to send](#selecting-to-send)
104 * [Selecting deferred values](#selecting-deferred-values)
105 * [Switch over a channel of deferred values](#switch-over-a-channel-of-deferred-values)
Roman Elizarov8db17332017-03-09 12:40:45 +0300106* [Further reading](#further-reading)
Roman Elizarovfa7723e2017-02-06 11:17:51 +0300107
Roman Elizarova5e653f2017-02-13 13:49:55 +0300108<!--- END_TOC -->
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300109
110## Coroutine basics
111
112This section covers basic coroutine concepts.
113
114### Your first coroutine
Roman Elizarov7deefb82017-01-31 10:33:17 +0300115
116Run the following code:
117
118```kotlin
119fun main(args: Array<String>) {
Roman Elizarov66f018c2017-09-29 21:39:03 +0300120 launch { // launch new coroutine
Roman Elizarov7deefb82017-01-31 10:33:17 +0300121 delay(1000L) // non-blocking delay for 1 second (default time unit is ms)
122 println("World!") // print after delay
123 }
124 println("Hello,") // main function continues while coroutine is delayed
125 Thread.sleep(2000L) // block main thread for 2 seconds to keep JVM alive
126}
127```
128
Roman Elizarove8d79342017-08-29 15:21:21 +0300129> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-basic-01.kt)
Roman Elizarov7deefb82017-01-31 10:33:17 +0300130
131Run this code:
132
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300133```text
Roman Elizarov7deefb82017-01-31 10:33:17 +0300134Hello,
135World!
136```
137
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300138<!--- TEST -->
139
Roman Elizarov419a6c82017-02-09 18:36:22 +0300140Essentially, coroutines are light-weight threads.
141They are launched with [launch] _coroutine builder_.
142You can achieve the same result replacing
Roman Elizarov66f018c2017-09-29 21:39:03 +0300143`launch { ... }` with `thread { ... }` and `delay(...)` with `Thread.sleep(...)`. Try it.
Roman Elizarov7deefb82017-01-31 10:33:17 +0300144
Roman Elizarov66f018c2017-09-29 21:39:03 +0300145If you start by replacing `launch` by `thread`, the compiler produces the following error:
Roman Elizarov7deefb82017-01-31 10:33:17 +0300146
147```
148Error: Kotlin: Suspend functions are only allowed to be called from a coroutine or another suspend function
149```
150
Roman Elizarov419a6c82017-02-09 18:36:22 +0300151That is because [delay] is a special _suspending function_ that does not block a thread, but _suspends_
Roman Elizarov7deefb82017-01-31 10:33:17 +0300152coroutine and it can be only used from a coroutine.
153
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300154### Bridging blocking and non-blocking worlds
Roman Elizarov7deefb82017-01-31 10:33:17 +0300155
156The first example mixes _non-blocking_ `delay(...)` and _blocking_ `Thread.sleep(...)` in the same
157code of `main` function. It is easy to get lost. Let's cleanly separate blocking and non-blocking
Roman Elizarov419a6c82017-02-09 18:36:22 +0300158worlds by using [runBlocking]:
Roman Elizarov7deefb82017-01-31 10:33:17 +0300159
160```kotlin
161fun main(args: Array<String>) = runBlocking<Unit> { // start main coroutine
Roman Elizarov66f018c2017-09-29 21:39:03 +0300162 launch { // launch new coroutine
Roman Elizarov7deefb82017-01-31 10:33:17 +0300163 delay(1000L)
164 println("World!")
165 }
166 println("Hello,") // main coroutine continues while child is delayed
167 delay(2000L) // non-blocking delay for 2 seconds to keep JVM alive
168}
169```
170
Roman Elizarove8d79342017-08-29 15:21:21 +0300171> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-basic-02.kt)
Roman Elizarov7deefb82017-01-31 10:33:17 +0300172
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300173<!--- TEST
174Hello,
175World!
176-->
177
Roman Elizarov419a6c82017-02-09 18:36:22 +0300178The result is the same, but this code uses only non-blocking [delay].
Roman Elizarov7deefb82017-01-31 10:33:17 +0300179
180`runBlocking { ... }` works as an adaptor that is used here to start the top-level main coroutine.
181The regular code outside of `runBlocking` _blocks_, until the coroutine inside `runBlocking` is active.
182
183This is also a way to write unit-tests for suspending functions:
184
185```kotlin
186class MyTest {
187 @Test
188 fun testMySuspendingFunction() = runBlocking<Unit> {
189 // here we can use suspending functions using any assertion style that we like
190 }
191}
192```
Roman Elizarovb3d55a52017-02-03 12:47:21 +0300193
194<!--- CLEAR -->
Roman Elizarov7deefb82017-01-31 10:33:17 +0300195
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300196### Waiting for a job
Roman Elizarov7deefb82017-01-31 10:33:17 +0300197
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300198Delaying for a time while another coroutine is working is not a good approach. Let's explicitly
Roman Elizarov419a6c82017-02-09 18:36:22 +0300199wait (in a non-blocking way) until the background [Job] that we have launched is complete:
Roman Elizarov7deefb82017-01-31 10:33:17 +0300200
201```kotlin
202fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov66f018c2017-09-29 21:39:03 +0300203 val job = launch { // launch new coroutine and keep a reference to its Job
Roman Elizarov7deefb82017-01-31 10:33:17 +0300204 delay(1000L)
205 println("World!")
206 }
207 println("Hello,")
208 job.join() // wait until child coroutine completes
209}
210```
211
Roman Elizarove8d79342017-08-29 15:21:21 +0300212> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-basic-03.kt)
Roman Elizarov7deefb82017-01-31 10:33:17 +0300213
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300214<!--- TEST
215Hello,
216World!
217-->
218
Roman Elizarov7deefb82017-01-31 10:33:17 +0300219Now the result is still the same, but the code of the main coroutine is not tied to the duration of
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300220the background job in any way. Much better.
Roman Elizarov7deefb82017-01-31 10:33:17 +0300221
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300222### Extract function refactoring
Roman Elizarov7deefb82017-01-31 10:33:17 +0300223
Roman Elizarov66f018c2017-09-29 21:39:03 +0300224Let's extract the block of code inside `launch { ... }` into a separate function. When you
Roman Elizarov7deefb82017-01-31 10:33:17 +0300225perform "Extract function" refactoring on this code you get a new function with `suspend` modifier.
226That is your first _suspending function_. Suspending functions can be used inside coroutines
227just like regular functions, but their additional feature is that they can, in turn,
228use other suspending functions, like `delay` in this example, to _suspend_ execution of a coroutine.
229
230```kotlin
231fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov66f018c2017-09-29 21:39:03 +0300232 val job = launch { doWorld() }
Roman Elizarov7deefb82017-01-31 10:33:17 +0300233 println("Hello,")
234 job.join()
235}
236
237// this is your first suspending function
238suspend fun doWorld() {
239 delay(1000L)
240 println("World!")
241}
242```
243
Roman Elizarove8d79342017-08-29 15:21:21 +0300244> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-basic-04.kt)
Roman Elizarov7deefb82017-01-31 10:33:17 +0300245
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300246<!--- TEST
247Hello,
248World!
249-->
250
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300251### Coroutines ARE light-weight
Roman Elizarov7deefb82017-01-31 10:33:17 +0300252
253Run the following code:
254
255```kotlin
256fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov66f018c2017-09-29 21:39:03 +0300257 val jobs = List(100_000) { // launch a lot of coroutines and list their jobs
258 launch {
Roman Elizarov7deefb82017-01-31 10:33:17 +0300259 delay(1000L)
260 print(".")
261 }
262 }
263 jobs.forEach { it.join() } // wait for all jobs to complete
264}
265```
266
Roman Elizarove8d79342017-08-29 15:21:21 +0300267> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-basic-05.kt)
Roman Elizarov7deefb82017-01-31 10:33:17 +0300268
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300269<!--- TEST lines.size == 1 && lines[0] == ".".repeat(100_000) -->
270
Roman Elizarov66f018c2017-09-29 21:39:03 +0300271It launches 100K coroutines and, after a second, each coroutine prints a dot.
Roman Elizarov7deefb82017-01-31 10:33:17 +0300272Now, try that with threads. What would happen? (Most likely your code will produce some sort of out-of-memory error)
273
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300274### Coroutines are like daemon threads
Roman Elizarov7deefb82017-01-31 10:33:17 +0300275
276The following code launches a long-running coroutine that prints "I'm sleeping" twice a second and then
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300277returns from the main function after some delay:
Roman Elizarov7deefb82017-01-31 10:33:17 +0300278
279```kotlin
280fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov66f018c2017-09-29 21:39:03 +0300281 launch {
Roman Elizarov7deefb82017-01-31 10:33:17 +0300282 repeat(1000) { i ->
283 println("I'm sleeping $i ...")
284 delay(500L)
285 }
286 }
287 delay(1300L) // just quit after delay
288}
289```
290
Roman Elizarove8d79342017-08-29 15:21:21 +0300291> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-basic-06.kt)
Roman Elizarov7deefb82017-01-31 10:33:17 +0300292
293You can run and see that it prints three lines and terminates:
294
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300295```text
Roman Elizarov7deefb82017-01-31 10:33:17 +0300296I'm sleeping 0 ...
297I'm sleeping 1 ...
298I'm sleeping 2 ...
299```
300
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300301<!--- TEST -->
302
Roman Elizarov7deefb82017-01-31 10:33:17 +0300303Active coroutines do not keep the process alive. They are like daemon threads.
304
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300305## Cancellation and timeouts
306
307This section covers coroutine cancellation and timeouts.
308
309### Cancelling coroutine execution
Roman Elizarov7deefb82017-01-31 10:33:17 +0300310
311In small application the return from "main" method might sound like a good idea to get all coroutines
312implicitly terminated. In a larger, long-running application, you need finer-grained control.
Roman Elizarov419a6c82017-02-09 18:36:22 +0300313The [launch] function returns a [Job] that can be used to cancel running coroutine:
Roman Elizarov7deefb82017-01-31 10:33:17 +0300314
315```kotlin
316fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov66f018c2017-09-29 21:39:03 +0300317 val job = launch {
Roman Elizarov7deefb82017-01-31 10:33:17 +0300318 repeat(1000) { i ->
319 println("I'm sleeping $i ...")
320 delay(500L)
321 }
322 }
323 delay(1300L) // delay a bit
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300324 println("main: I'm tired of waiting!")
Roman Elizarov7deefb82017-01-31 10:33:17 +0300325 job.cancel() // cancels the job
Roman Elizarov8b38fa22017-09-27 17:44:31 +0300326 job.join() // waits for job's completion
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300327 println("main: Now I can quit.")
Roman Elizarov7deefb82017-01-31 10:33:17 +0300328}
329```
330
Roman Elizarove8d79342017-08-29 15:21:21 +0300331> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-cancel-01.kt)
Roman Elizarov7deefb82017-01-31 10:33:17 +0300332
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300333It produces the following output:
334
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300335```text
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300336I'm sleeping 0 ...
337I'm sleeping 1 ...
338I'm sleeping 2 ...
339main: I'm tired of waiting!
340main: Now I can quit.
341```
342
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300343<!--- TEST -->
344
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300345As soon as main invokes `job.cancel`, we don't see any output from the other coroutine because it was cancelled.
Roman Elizarov88396732017-09-27 21:30:47 +0300346There is also a [Job] extension function [cancelAndJoin]
347that combines [cancel][Job.cancel] and [join][Job.join] invocations.
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300348
349### Cancellation is cooperative
Roman Elizarov7deefb82017-01-31 10:33:17 +0300350
Tair Rzayevaf734622017-02-01 22:30:16 +0200351Coroutine cancellation is _cooperative_. A coroutine code has to cooperate to be cancellable.
Roman Elizarov7deefb82017-01-31 10:33:17 +0300352All the suspending functions in `kotlinx.coroutines` are _cancellable_. They check for cancellation of
Roman Elizarov419a6c82017-02-09 18:36:22 +0300353coroutine and throw [CancellationException] when cancelled. However, if a coroutine is working in
Roman Elizarov7deefb82017-01-31 10:33:17 +0300354a computation and does not check for cancellation, then it cannot be cancelled, like the following
355example shows:
356
357```kotlin
358fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov24cd6542017-08-03 21:20:04 -0700359 val startTime = System.currentTimeMillis()
Roman Elizarov66f018c2017-09-29 21:39:03 +0300360 val job = launch {
Roman Elizarov24cd6542017-08-03 21:20:04 -0700361 var nextPrintTime = startTime
Roman Elizarov7deefb82017-01-31 10:33:17 +0300362 var i = 0
Roman Elizarov8b38fa22017-09-27 17:44:31 +0300363 while (i < 5) { // computation loop, just wastes CPU
Roman Elizarov24cd6542017-08-03 21:20:04 -0700364 // print a message twice a second
365 if (System.currentTimeMillis() >= nextPrintTime) {
Roman Elizarov7deefb82017-01-31 10:33:17 +0300366 println("I'm sleeping ${i++} ...")
Roman Elizarov35d2c342017-07-20 14:54:39 +0300367 nextPrintTime += 500L
Roman Elizarov7deefb82017-01-31 10:33:17 +0300368 }
369 }
370 }
371 delay(1300L) // delay a bit
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300372 println("main: I'm tired of waiting!")
Roman Elizarov8b38fa22017-09-27 17:44:31 +0300373 job.cancelAndJoin() // cancels the job and waits for its completion
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300374 println("main: Now I can quit.")
Roman Elizarov7deefb82017-01-31 10:33:17 +0300375}
376```
377
Roman Elizarove8d79342017-08-29 15:21:21 +0300378> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-cancel-02.kt)
Roman Elizarov7deefb82017-01-31 10:33:17 +0300379
Roman Elizarov8b38fa22017-09-27 17:44:31 +0300380Run it to see that it continues to print "I'm sleeping" even after cancellation
381until the job completes by itself after five iterations.
Roman Elizarov7deefb82017-01-31 10:33:17 +0300382
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300383<!--- TEST
384I'm sleeping 0 ...
385I'm sleeping 1 ...
386I'm sleeping 2 ...
387main: I'm tired of waiting!
388I'm sleeping 3 ...
389I'm sleeping 4 ...
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300390main: Now I can quit.
391-->
392
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300393### Making computation code cancellable
Roman Elizarov7deefb82017-01-31 10:33:17 +0300394
395There are two approaches to making computation code cancellable. The first one is to periodically
Roman Elizarov66f018c2017-09-29 21:39:03 +0300396invoke a suspending function that checks for cancellation. There is a [yield] function that is a good choice for that purpose.
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300397The other one is to explicitly check the cancellation status. Let us try the later approach.
Roman Elizarov7deefb82017-01-31 10:33:17 +0300398
Roman Elizarov8b38fa22017-09-27 17:44:31 +0300399Replace `while (i < 5)` in the previous example with `while (isActive)` and rerun it.
Roman Elizarov7deefb82017-01-31 10:33:17 +0300400
Roman Elizarovb3d55a52017-02-03 12:47:21 +0300401```kotlin
402fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov24cd6542017-08-03 21:20:04 -0700403 val startTime = System.currentTimeMillis()
Roman Elizarov66f018c2017-09-29 21:39:03 +0300404 val job = launch {
Roman Elizarov24cd6542017-08-03 21:20:04 -0700405 var nextPrintTime = startTime
Roman Elizarovb3d55a52017-02-03 12:47:21 +0300406 var i = 0
407 while (isActive) { // cancellable computation loop
Roman Elizarov24cd6542017-08-03 21:20:04 -0700408 // print a message twice a second
409 if (System.currentTimeMillis() >= nextPrintTime) {
Roman Elizarovb3d55a52017-02-03 12:47:21 +0300410 println("I'm sleeping ${i++} ...")
Roman Elizarov24cd6542017-08-03 21:20:04 -0700411 nextPrintTime += 500L
Roman Elizarovb3d55a52017-02-03 12:47:21 +0300412 }
413 }
414 }
415 delay(1300L) // delay a bit
416 println("main: I'm tired of waiting!")
Roman Elizarov8b38fa22017-09-27 17:44:31 +0300417 job.cancelAndJoin() // cancels the job and waits for its completion
Roman Elizarovb3d55a52017-02-03 12:47:21 +0300418 println("main: Now I can quit.")
419}
420```
421
Roman Elizarove8d79342017-08-29 15:21:21 +0300422> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-cancel-03.kt)
Roman Elizarov7deefb82017-01-31 10:33:17 +0300423
Roman Elizarov8b38fa22017-09-27 17:44:31 +0300424As you can see, now this loop is cancelled. [isActive][CoroutineScope.isActive] is a property that is available inside
Roman Elizarov419a6c82017-02-09 18:36:22 +0300425the code of coroutines via [CoroutineScope] object.
Roman Elizarov7deefb82017-01-31 10:33:17 +0300426
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300427<!--- TEST
428I'm sleeping 0 ...
429I'm sleeping 1 ...
430I'm sleeping 2 ...
431main: I'm tired of waiting!
432main: Now I can quit.
433-->
434
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300435### Closing resources with finally
436
Roman Elizarov419a6c82017-02-09 18:36:22 +0300437Cancellable suspending functions throw [CancellationException] on cancellation which can be handled in
Roman Elizarov66f018c2017-09-29 21:39:03 +0300438all the usual way. For example, `try {...} finally {...}` expression and Kotlin `use` function execute their
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300439finalization actions normally when coroutine is cancelled:
440
441```kotlin
442fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov66f018c2017-09-29 21:39:03 +0300443 val job = launch {
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300444 try {
445 repeat(1000) { i ->
446 println("I'm sleeping $i ...")
447 delay(500L)
448 }
449 } finally {
450 println("I'm running finally")
451 }
452 }
453 delay(1300L) // delay a bit
454 println("main: I'm tired of waiting!")
Roman Elizarov8b38fa22017-09-27 17:44:31 +0300455 job.cancelAndJoin() // cancels the job and waits for its completion
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300456 println("main: Now I can quit.")
457}
458```
459
Roman Elizarove8d79342017-08-29 15:21:21 +0300460> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-cancel-04.kt)
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300461
Roman Elizarov88396732017-09-27 21:30:47 +0300462Both [join][Job.join] and [cancelAndJoin] wait for all the finalization actions to complete,
463so the example above produces the following output:
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300464
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300465```text
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300466I'm sleeping 0 ...
467I'm sleeping 1 ...
468I'm sleeping 2 ...
469main: I'm tired of waiting!
470I'm running finally
471main: Now I can quit.
472```
473
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300474<!--- TEST -->
475
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300476### Run non-cancellable block
477
478Any attempt to use a suspending function in the `finally` block of the previous example will cause
Roman Elizarov419a6c82017-02-09 18:36:22 +0300479[CancellationException], because the coroutine running this code is cancelled. Usually, this is not a
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300480problem, since all well-behaving closing operations (closing a file, cancelling a job, or closing any kind of a
481communication channel) are usually non-blocking and do not involve any suspending functions. However, in the
482rare case when you need to suspend in the cancelled coroutine you can wrap the corresponding code in
Roman Elizarov419a6c82017-02-09 18:36:22 +0300483`run(NonCancellable) {...}` using [run] function and [NonCancellable] context as the following example shows:
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300484
485```kotlin
486fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov66f018c2017-09-29 21:39:03 +0300487 val job = launch {
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300488 try {
489 repeat(1000) { i ->
490 println("I'm sleeping $i ...")
491 delay(500L)
492 }
493 } finally {
494 run(NonCancellable) {
495 println("I'm running finally")
496 delay(1000L)
497 println("And I've just delayed for 1 sec because I'm non-cancellable")
498 }
499 }
500 }
501 delay(1300L) // delay a bit
502 println("main: I'm tired of waiting!")
Roman Elizarov8b38fa22017-09-27 17:44:31 +0300503 job.cancelAndJoin() // cancels the job and waits for its completion
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300504 println("main: Now I can quit.")
505}
506```
507
Roman Elizarove8d79342017-08-29 15:21:21 +0300508> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-cancel-05.kt)
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300509
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300510<!--- TEST
511I'm sleeping 0 ...
512I'm sleeping 1 ...
513I'm sleeping 2 ...
514main: I'm tired of waiting!
515I'm running finally
516And I've just delayed for 1 sec because I'm non-cancellable
517main: Now I can quit.
518-->
519
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300520### Timeout
521
522The most obvious reason to cancel coroutine execution in practice,
523is because its execution time has exceeded some timeout.
Roman Elizarov419a6c82017-02-09 18:36:22 +0300524While you can manually track the reference to the corresponding [Job] and launch a separate coroutine to cancel
525the tracked one after delay, there is a ready to use [withTimeout] function that does it.
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300526Look at the following example:
527
528```kotlin
529fun main(args: Array<String>) = runBlocking<Unit> {
530 withTimeout(1300L) {
531 repeat(1000) { i ->
532 println("I'm sleeping $i ...")
533 delay(500L)
534 }
535 }
536}
537```
538
Roman Elizarove8d79342017-08-29 15:21:21 +0300539> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-cancel-06.kt)
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300540
541It produces the following output:
542
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300543```text
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300544I'm sleeping 0 ...
545I'm sleeping 1 ...
546I'm sleeping 2 ...
Roman Elizarov63f6ea22017-09-06 18:42:34 +0300547Exception in thread "main" kotlinx.coroutines.experimental.TimeoutCancellationException: Timed out waiting for 1300 MILLISECONDS
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300548```
549
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300550<!--- TEST STARTS_WITH -->
551
Roman Elizarov63f6ea22017-09-06 18:42:34 +0300552The `TimeoutCancellationException` that is thrown by [withTimeout] is a subclass of [CancellationException].
Roman Elizarovca9d5be2017-04-20 19:23:18 +0300553We have not seen its stack trace printed on the console before. That is because
Roman Elizarov7c864d82017-02-27 10:17:50 +0300554inside a cancelled coroutine `CancellationException` is considered to be a normal reason for coroutine completion.
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300555However, in this example we have used `withTimeout` right inside the `main` function.
556
557Because cancellation is just an exception, all the resources will be closed in a usual way.
Roman Elizarov63f6ea22017-09-06 18:42:34 +0300558You can wrap the code with timeout in `try {...} catch (e: TimeoutCancellationException) {...}` block if
559you need to do some additional action specifically on any kind of timeout or use [withTimeoutOrNull] function
Roman Elizarov8b38fa22017-09-27 17:44:31 +0300560that is similar to [withTimeout], but returns `null` on timeout instead of throwing an exception:
561
562```kotlin
563fun main(args: Array<String>) = runBlocking<Unit> {
564 val result = withTimeoutOrNull(1300L) {
565 repeat(1000) { i ->
566 println("I'm sleeping $i ...")
567 delay(500L)
568 }
569 "Done" // will get cancelled before it produces this result
570 }
571 println("Result is $result")
572}
573```
574
575> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-cancel-07.kt)
576
577There is no longer an exception when running this code:
578
579```text
580I'm sleeping 0 ...
581I'm sleeping 1 ...
582I'm sleeping 2 ...
583Result is null
584```
585
586<!--- TEST -->
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300587
588## Composing suspending functions
589
590This section covers various approaches to composition of suspending functions.
591
592### Sequential by default
593
594Assume that we have two suspending functions defined elsewhere that do something useful like some kind of
Roman Elizarovb7721cf2017-02-03 19:23:08 +0300595remote service call or computation. We just pretend they are useful, but actually each one just
596delays for a second for the purpose of this example:
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300597
Roman Elizarovfa7723e2017-02-06 11:17:51 +0300598<!--- INCLUDE .*/example-compose-([0-9]+).kt
599import kotlin.system.measureTimeMillis
600-->
601
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300602```kotlin
603suspend fun doSomethingUsefulOne(): Int {
604 delay(1000L) // pretend we are doing something useful here
605 return 13
606}
607
608suspend fun doSomethingUsefulTwo(): Int {
609 delay(1000L) // pretend we are doing something useful here, too
610 return 29
611}
612```
613
Roman Elizarovfa7723e2017-02-06 11:17:51 +0300614<!--- INCLUDE .*/example-compose-([0-9]+).kt -->
615
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300616What do we do if need to invoke them _sequentially_ -- first `doSomethingUsefulOne` _and then_
617`doSomethingUsefulTwo` and compute the sum of their results?
618In practise we do this if we use the results of the first function to make a decision on whether we need
619to invoke the second one or to decide on how to invoke it.
620
621We just use a normal sequential invocation, because the code in the coroutine, just like in the regular
Roman Elizarov32d95322017-02-09 15:57:31 +0300622code, is _sequential_ by default. The following example demonstrates it by measuring the total
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300623time it takes to execute both suspending functions:
624
625```kotlin
626fun main(args: Array<String>) = runBlocking<Unit> {
627 val time = measureTimeMillis {
628 val one = doSomethingUsefulOne()
629 val two = doSomethingUsefulTwo()
630 println("The answer is ${one + two}")
631 }
632 println("Completed in $time ms")
633}
634```
635
Roman Elizarove8d79342017-08-29 15:21:21 +0300636> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-compose-01.kt)
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300637
638It produces something like this:
639
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300640```text
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300641The answer is 42
642Completed in 2017 ms
643```
644
Roman Elizarov35d2c342017-07-20 14:54:39 +0300645<!--- TEST ARBITRARY_TIME -->
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300646
Roman Elizarov32d95322017-02-09 15:57:31 +0300647### Concurrent using async
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300648
649What if there are no dependencies between invocation of `doSomethingUsefulOne` and `doSomethingUsefulTwo` and
Roman Elizarov419a6c82017-02-09 18:36:22 +0300650we want to get the answer faster, by doing both _concurrently_? This is where [async] comes to help.
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300651
Roman Elizarov419a6c82017-02-09 18:36:22 +0300652Conceptually, [async] is just like [launch]. It starts a separate coroutine which is a light-weight thread
653that works concurrently with all the other coroutines. The difference is that `launch` returns a [Job] and
654does not carry any resulting value, while `async` returns a [Deferred] -- a light-weight non-blocking future
Roman Elizarov32d95322017-02-09 15:57:31 +0300655that represents a promise to provide a result later. You can use `.await()` on a deferred value to get its eventual result,
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300656but `Deferred` is also a `Job`, so you can cancel it if needed.
657
658```kotlin
659fun main(args: Array<String>) = runBlocking<Unit> {
660 val time = measureTimeMillis {
Roman Elizarov66f018c2017-09-29 21:39:03 +0300661 val one = async { doSomethingUsefulOne() }
662 val two = async { doSomethingUsefulTwo() }
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300663 println("The answer is ${one.await() + two.await()}")
664 }
665 println("Completed in $time ms")
666}
667```
668
Roman Elizarove8d79342017-08-29 15:21:21 +0300669> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-compose-02.kt)
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300670
671It produces something like this:
672
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300673```text
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300674The answer is 42
675Completed in 1017 ms
676```
677
Roman Elizarov35d2c342017-07-20 14:54:39 +0300678<!--- TEST ARBITRARY_TIME -->
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300679
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300680This is twice as fast, because we have concurrent execution of two coroutines.
681Note, that concurrency with coroutines is always explicit.
682
Roman Elizarov32d95322017-02-09 15:57:31 +0300683### Lazily started async
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300684
Roman Elizarov66f018c2017-09-29 21:39:03 +0300685There is a laziness option to [async] using an optional `start` parameter with a value of [CoroutineStart.LAZY].
Roman Elizarov419a6c82017-02-09 18:36:22 +0300686It starts coroutine only when its result is needed by some
687[await][Deferred.await] or if a [start][Job.start] function
Roman Elizarov32d95322017-02-09 15:57:31 +0300688is invoked. Run the following example that differs from the previous one only by this option:
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300689
690```kotlin
691fun main(args: Array<String>) = runBlocking<Unit> {
692 val time = measureTimeMillis {
Roman Elizarov66f018c2017-09-29 21:39:03 +0300693 val one = async(start = CoroutineStart.LAZY) { doSomethingUsefulOne() }
694 val two = async(start = CoroutineStart.LAZY) { doSomethingUsefulTwo() }
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300695 println("The answer is ${one.await() + two.await()}")
696 }
697 println("Completed in $time ms")
698}
699```
700
Roman Elizarove8d79342017-08-29 15:21:21 +0300701> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-compose-03.kt)
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300702
703It produces something like this:
704
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300705```text
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300706The answer is 42
707Completed in 2017 ms
708```
709
Roman Elizarov35d2c342017-07-20 14:54:39 +0300710<!--- TEST ARBITRARY_TIME -->
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300711
Roman Elizarov32d95322017-02-09 15:57:31 +0300712So, we are back to sequential execution, because we _first_ start and await for `one`, _and then_ start and await
713for `two`. It is not the intended use-case for laziness. It is designed as a replacement for
714the standard `lazy` function in cases when computation of the value involves suspending functions.
715
716### Async-style functions
717
718We can define async-style functions that invoke `doSomethingUsefulOne` and `doSomethingUsefulTwo`
Roman Elizarov419a6c82017-02-09 18:36:22 +0300719_asynchronously_ using [async] coroutine builder. It is a good style to name such functions with
Roman Elizarov32d95322017-02-09 15:57:31 +0300720either "async" prefix of "Async" suffix to highlight the fact that they only start asynchronous
721computation and one needs to use the resulting deferred value to get the result.
722
723```kotlin
724// The result type of asyncSomethingUsefulOne is Deferred<Int>
Roman Elizarov66f018c2017-09-29 21:39:03 +0300725fun asyncSomethingUsefulOne() = async {
Roman Elizarov32d95322017-02-09 15:57:31 +0300726 doSomethingUsefulOne()
727}
728
729// The result type of asyncSomethingUsefulTwo is Deferred<Int>
Roman Elizarov66f018c2017-09-29 21:39:03 +0300730fun asyncSomethingUsefulTwo() = async {
Roman Elizarov32d95322017-02-09 15:57:31 +0300731 doSomethingUsefulTwo()
732}
733```
734
735Note, that these `asyncXXX` function are **not** _suspending_ functions. They can be used from anywhere.
736However, their use always implies asynchronous (here meaning _concurrent_) execution of their action
737with the invoking code.
738
739The following example shows their use outside of coroutine:
740
741```kotlin
742// note, that we don't have `runBlocking` to the right of `main` in this example
743fun main(args: Array<String>) {
744 val time = measureTimeMillis {
745 // we can initiate async actions outside of a coroutine
746 val one = asyncSomethingUsefulOne()
747 val two = asyncSomethingUsefulTwo()
748 // but waiting for a result must involve either suspending or blocking.
749 // here we use `runBlocking { ... }` to block the main thread while waiting for the result
750 runBlocking {
751 println("The answer is ${one.await() + two.await()}")
752 }
753 }
754 println("Completed in $time ms")
755}
756```
757
Roman Elizarove8d79342017-08-29 15:21:21 +0300758> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-compose-04.kt)
Roman Elizarov1293ccd2017-02-01 18:49:54 +0300759
Roman Elizarov35d2c342017-07-20 14:54:39 +0300760<!--- TEST ARBITRARY_TIME
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300761The answer is 42
762Completed in 1085 ms
763-->
764
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300765## Coroutine context and dispatchers
766
Roman Elizarov66f018c2017-09-29 21:39:03 +0300767Coroutines always execute in some context which is represented by the value of
768[CoroutineContext](https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.coroutines.experimental/-coroutine-context/)
769type, defined in the Kotlin standard library.
770
771The coroutine context is a set of various elements. The main elements are the [Job] of the coroutine,
772which we've seen before, and its dispatcher, which is covered in this section.
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300773
774### Dispatchers and threads
775
Roman Elizarov66f018c2017-09-29 21:39:03 +0300776Coroutine context includes a _coroutine dispatcher_ (see [CoroutineDispatcher]) that determines what thread or threads
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300777the corresponding coroutine uses for its execution. Coroutine dispatcher can confine coroutine execution
Roman Elizarov66f018c2017-09-29 21:39:03 +0300778to a specific thread, dispatch it to a thread pool, or let it run unconfined.
779
780All coroutines builders like [launch] and [async] accept an optional
781[CoroutineContext](https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.coroutines.experimental/-coroutine-context/)
782parameter that can be used to explicitly specify the dispatcher for new coroutine and other context elements.
783
784Try the following example:
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300785
786```kotlin
787fun main(args: Array<String>) = runBlocking<Unit> {
788 val jobs = arrayListOf<Job>()
789 jobs += launch(Unconfined) { // not confined -- will work with main thread
Roman Elizarov43e3af72017-07-21 16:01:31 +0300790 println(" 'Unconfined': I'm working in thread ${Thread.currentThread().name}")
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300791 }
Roman Elizarov43e3af72017-07-21 16:01:31 +0300792 jobs += launch(coroutineContext) { // context of the parent, runBlocking coroutine
793 println("'coroutineContext': I'm working in thread ${Thread.currentThread().name}")
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300794 }
795 jobs += launch(CommonPool) { // will get dispatched to ForkJoinPool.commonPool (or equivalent)
Roman Elizarov43e3af72017-07-21 16:01:31 +0300796 println(" 'CommonPool': I'm working in thread ${Thread.currentThread().name}")
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300797 }
798 jobs += launch(newSingleThreadContext("MyOwnThread")) { // will get its own new thread
Roman Elizarov43e3af72017-07-21 16:01:31 +0300799 println(" 'newSTC': I'm working in thread ${Thread.currentThread().name}")
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300800 }
801 jobs.forEach { it.join() }
802}
803```
804
Roman Elizarove8d79342017-08-29 15:21:21 +0300805> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-context-01.kt)
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300806
807It produces the following output (maybe in different order):
808
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300809```text
Roman Elizarov43e3af72017-07-21 16:01:31 +0300810 'Unconfined': I'm working in thread main
811 'CommonPool': I'm working in thread ForkJoinPool.commonPool-worker-1
812 'newSTC': I'm working in thread MyOwnThread
813'coroutineContext': I'm working in thread main
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300814```
815
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300816<!--- TEST LINES_START_UNORDERED -->
817
Roman Elizarov66f018c2017-09-29 21:39:03 +0300818The default dispatcher that we've used in previous sections is representend by [DefaultDispather], which
819is equal to [CommonPool] in the current implementation. So, `launch { ... }` is the same
820as `launch(DefaultDispather) { ... }`, which is the same as `launch(CommonPool) { ... }`.
821
Roman Elizarov43e3af72017-07-21 16:01:31 +0300822The difference between parent [coroutineContext][CoroutineScope.coroutineContext] and
823[Unconfined] context will be shown later.
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300824
825### Unconfined vs confined dispatcher
826
Roman Elizarov419a6c82017-02-09 18:36:22 +0300827The [Unconfined] coroutine dispatcher starts coroutine in the caller thread, but only until the
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300828first suspension point. After suspension it resumes in the thread that is fully determined by the
829suspending function that was invoked. Unconfined dispatcher is appropriate when coroutine does not
830consume CPU time nor updates any shared data (like UI) that is confined to a specific thread.
831
Roman Elizarov43e3af72017-07-21 16:01:31 +0300832On the other side, [coroutineContext][CoroutineScope.coroutineContext] property that is available inside the block of any coroutine
Roman Elizarov419a6c82017-02-09 18:36:22 +0300833via [CoroutineScope] interface, is a reference to a context of this particular coroutine.
Roman Elizarov66f018c2017-09-29 21:39:03 +0300834This way, a parent context can be inherited. The default dispatcher for [runBlocking] coroutine, in particular,
835is confined to the invoker thread, so inheriting it has the effect of confining execution to
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300836this thread with a predictable FIFO scheduling.
837
838```kotlin
839fun main(args: Array<String>) = runBlocking<Unit> {
840 val jobs = arrayListOf<Job>()
841 jobs += launch(Unconfined) { // not confined -- will work with main thread
Roman Elizarov43e3af72017-07-21 16:01:31 +0300842 println(" 'Unconfined': I'm working in thread ${Thread.currentThread().name}")
Roman Elizarovd0021622017-03-10 15:43:38 +0300843 delay(500)
Roman Elizarov43e3af72017-07-21 16:01:31 +0300844 println(" 'Unconfined': After delay in thread ${Thread.currentThread().name}")
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300845 }
Roman Elizarov43e3af72017-07-21 16:01:31 +0300846 jobs += launch(coroutineContext) { // context of the parent, runBlocking coroutine
847 println("'coroutineContext': I'm working in thread ${Thread.currentThread().name}")
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300848 delay(1000)
Roman Elizarov43e3af72017-07-21 16:01:31 +0300849 println("'coroutineContext': After delay in thread ${Thread.currentThread().name}")
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300850 }
851 jobs.forEach { it.join() }
852}
853```
854
Roman Elizarove8d79342017-08-29 15:21:21 +0300855> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-context-02.kt)
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300856
857Produces the output:
858
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300859```text
Roman Elizarov43e3af72017-07-21 16:01:31 +0300860 'Unconfined': I'm working in thread main
861'coroutineContext': I'm working in thread main
862 'Unconfined': After delay in thread kotlinx.coroutines.DefaultExecutor
863'coroutineContext': After delay in thread main
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300864```
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300865
866<!--- TEST LINES_START -->
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300867
Roman Elizarov43e3af72017-07-21 16:01:31 +0300868So, the coroutine that had inherited `coroutineContext` of `runBlocking {...}` continues to execute
869in the `main` thread, while the unconfined one had resumed in the default executor thread that [delay]
870function is using.
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300871
872### Debugging coroutines and threads
873
Roman Elizarov419a6c82017-02-09 18:36:22 +0300874Coroutines can suspend on one thread and resume on another thread with [Unconfined] dispatcher or
Roman Elizarov66f018c2017-09-29 21:39:03 +0300875with a default multi-threaded dispatcher. Even with a single-threaded dispatcher it might be hard to
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300876figure out what coroutine was doing what, where, and when. The common approach to debugging applications with
877threads is to print the thread name in the log file on each log statement. This feature is universally supported
878by logging frameworks. When using coroutines, the thread name alone does not give much of a context, so
879`kotlinx.coroutines` includes debugging facilities to make it easier.
880
881Run the following code with `-Dkotlinx.coroutines.debug` JVM option:
882
883```kotlin
884fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
885
886fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov43e3af72017-07-21 16:01:31 +0300887 val a = async(coroutineContext) {
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300888 log("I'm computing a piece of the answer")
889 6
890 }
Roman Elizarov43e3af72017-07-21 16:01:31 +0300891 val b = async(coroutineContext) {
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300892 log("I'm computing another piece of the answer")
893 7
894 }
895 log("The answer is ${a.await() * b.await()}")
896}
897```
898
Roman Elizarove8d79342017-08-29 15:21:21 +0300899> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-context-03.kt)
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300900
Roman Elizarovb7721cf2017-02-03 19:23:08 +0300901There are three coroutines. The main coroutine (#1) -- `runBlocking` one,
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300902and two coroutines computing deferred values `a` (#2) and `b` (#3).
903They are all executing in the context of `runBlocking` and are confined to the main thread.
904The output of this code is:
905
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300906```text
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300907[main @coroutine#2] I'm computing a piece of the answer
908[main @coroutine#3] I'm computing another piece of the answer
909[main @coroutine#1] The answer is 42
910```
911
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300912<!--- TEST -->
913
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300914The `log` function prints the name of the thread in square brackets and you can see, that it is the `main`
915thread, but the identifier of the currently executing coroutine is appended to it. This identifier
916is consecutively assigned to all created coroutines when debugging mode is turned on.
917
Roman Elizarov419a6c82017-02-09 18:36:22 +0300918You can read more about debugging facilities in the documentation for [newCoroutineContext] function.
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300919
920### Jumping between threads
921
922Run the following code with `-Dkotlinx.coroutines.debug` JVM option:
923
924```kotlin
925fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
926
927fun main(args: Array<String>) {
928 val ctx1 = newSingleThreadContext("Ctx1")
929 val ctx2 = newSingleThreadContext("Ctx2")
930 runBlocking(ctx1) {
931 log("Started in ctx1")
932 run(ctx2) {
933 log("Working in ctx2")
934 }
935 log("Back to ctx1")
936 }
937}
938```
939
Roman Elizarove8d79342017-08-29 15:21:21 +0300940> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-context-04.kt)
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300941
Roman Elizarov419a6c82017-02-09 18:36:22 +0300942It demonstrates two new techniques. One is using [runBlocking] with an explicitly specified context, and
943the second one is using [run] function to change a context of a coroutine while still staying in the
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300944same coroutine as you can see in the output below:
945
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300946```text
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300947[Ctx1 @coroutine#1] Started in ctx1
948[Ctx2 @coroutine#1] Working in ctx2
949[Ctx1 @coroutine#1] Back to ctx1
950```
951
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300952<!--- TEST -->
953
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300954### Job in the context
955
Roman Elizarov66f018c2017-09-29 21:39:03 +0300956The coroutine's [Job] is part of its context. The coroutine can retrieve it from its own context
Roman Elizarov43e3af72017-07-21 16:01:31 +0300957using `coroutineContext[Job]` expression:
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300958
959```kotlin
960fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov43e3af72017-07-21 16:01:31 +0300961 println("My job is ${coroutineContext[Job]}")
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300962}
963```
964
Roman Elizarove8d79342017-08-29 15:21:21 +0300965> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-context-05.kt)
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300966
Roman Elizarov66f018c2017-09-29 21:39:03 +0300967It produces something like that when running in [debug mode](#debugging-coroutines-and-threads):
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300968
969```
Roman Elizarov8b38fa22017-09-27 17:44:31 +0300970My job is "coroutine#1":BlockingCoroutine{Active}@6d311334
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300971```
972
Roman Elizarov8b38fa22017-09-27 17:44:31 +0300973<!--- TEST lines.size == 1 && lines[0].startsWith("My job is \"coroutine#1\":BlockingCoroutine{Active}@") -->
Roman Elizarov731f0ad2017-02-22 20:48:45 +0300974
Roman Elizarov43e3af72017-07-21 16:01:31 +0300975So, [isActive][CoroutineScope.isActive] in [CoroutineScope] is just a convenient shortcut for
976`coroutineContext[Job]!!.isActive`.
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300977
978### Children of a coroutine
979
Roman Elizarov43e3af72017-07-21 16:01:31 +0300980When [coroutineContext][CoroutineScope.coroutineContext] of a coroutine is used to launch another coroutine,
Roman Elizarov419a6c82017-02-09 18:36:22 +0300981the [Job] of the new coroutine becomes
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300982a _child_ of the parent coroutine's job. When the parent coroutine is cancelled, all its children
983are recursively cancelled, too.
984
985```kotlin
986fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov66f018c2017-09-29 21:39:03 +0300987 // launch a coroutine to process some kind of incoming request
988 val request = launch {
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300989 // it spawns two other jobs, one with its separate context
Roman Elizarov66f018c2017-09-29 21:39:03 +0300990 val job1 = launch {
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300991 println("job1: I have my own context and execute independently!")
992 delay(1000)
993 println("job1: I am not affected by cancellation of the request")
994 }
995 // and the other inherits the parent context
Roman Elizarov43e3af72017-07-21 16:01:31 +0300996 val job2 = launch(coroutineContext) {
Roman Elizarov2f6d7c92017-02-03 15:16:07 +0300997 println("job2: I am a child of the request coroutine")
998 delay(1000)
999 println("job2: I will not execute this line if my parent request is cancelled")
1000 }
1001 // request completes when both its sub-jobs complete:
1002 job1.join()
1003 job2.join()
1004 }
1005 delay(500)
1006 request.cancel() // cancel processing of the request
1007 delay(1000) // delay a second to see what happens
1008 println("main: Who has survived request cancellation?")
1009}
1010```
1011
Roman Elizarove8d79342017-08-29 15:21:21 +03001012> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-context-06.kt)
Roman Elizarov2f6d7c92017-02-03 15:16:07 +03001013
1014The output of this code is:
1015
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001016```text
Roman Elizarov2f6d7c92017-02-03 15:16:07 +03001017job1: I have my own context and execute independently!
1018job2: I am a child of the request coroutine
1019job1: I am not affected by cancellation of the request
1020main: Who has survived request cancellation?
1021```
1022
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001023<!--- TEST -->
1024
Roman Elizarov2f6d7c92017-02-03 15:16:07 +03001025### Combining contexts
1026
Roman Elizarov66f018c2017-09-29 21:39:03 +03001027Coroutine contexts can be combined using `+` operator. The context on the right-hand side replaces relevant entries
Roman Elizarov419a6c82017-02-09 18:36:22 +03001028of the context on the left-hand side. For example, a [Job] of the parent coroutine can be inherited, while
Roman Elizarov2f6d7c92017-02-03 15:16:07 +03001029its dispatcher replaced:
1030
1031```kotlin
1032fun main(args: Array<String>) = runBlocking<Unit> {
1033 // start a coroutine to process some kind of incoming request
Roman Elizarov43e3af72017-07-21 16:01:31 +03001034 val request = launch(coroutineContext) { // use the context of `runBlocking`
Roman Elizarov2f6d7c92017-02-03 15:16:07 +03001035 // spawns CPU-intensive child job in CommonPool !!!
Roman Elizarov43e3af72017-07-21 16:01:31 +03001036 val job = launch(coroutineContext + CommonPool) {
Roman Elizarov2f6d7c92017-02-03 15:16:07 +03001037 println("job: I am a child of the request coroutine, but with a different dispatcher")
1038 delay(1000)
1039 println("job: I will not execute this line if my parent request is cancelled")
1040 }
1041 job.join() // request completes when its sub-job completes
1042 }
1043 delay(500)
1044 request.cancel() // cancel processing of the request
1045 delay(1000) // delay a second to see what happens
1046 println("main: Who has survived request cancellation?")
1047}
1048```
1049
Roman Elizarove8d79342017-08-29 15:21:21 +03001050> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-context-07.kt)
Roman Elizarov2f6d7c92017-02-03 15:16:07 +03001051
1052The expected outcome of this code is:
1053
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001054```text
Roman Elizarov2f6d7c92017-02-03 15:16:07 +03001055job: I am a child of the request coroutine, but with a different dispatcher
1056main: Who has survived request cancellation?
1057```
1058
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001059<!--- TEST -->
1060
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001061### Parental responsibilities
1062
1063A parent coroutine always waits for completion of all its children. Parent does not have to explicitly track
Roman Elizarov88396732017-09-27 21:30:47 +03001064all the children it launches and it does not have to use [Job.join] to wait for them at the end:
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001065
1066```kotlin
1067fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov66f018c2017-09-29 21:39:03 +03001068 // launch a coroutine to process some kind of incoming request
1069 val request = launch {
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001070 repeat(3) { i -> // launch a few children jobs
1071 launch(coroutineContext) {
1072 delay((i + 1) * 200L) // variable delay 200ms, 400ms, 600ms
1073 println("Coroutine $i is done")
1074 }
1075 }
1076 println("request: I'm done and I don't explicitly join my children that are still active")
1077 }
1078 request.join() // wait for completion of the request, including all its children
1079 println("Now processing of the request is complete")
1080}
1081```
1082
1083> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-context-08.kt)
1084
1085The result is going to be:
1086
1087```text
1088request: I'm done and I don't explicitly join my children that are still active
1089Coroutine 0 is done
1090Coroutine 1 is done
1091Coroutine 2 is done
1092Now processing of the request is complete
1093```
1094
1095<!--- TEST -->
1096
Roman Elizarov2f6d7c92017-02-03 15:16:07 +03001097### Naming coroutines for debugging
1098
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001099Automatically assigned ids are good when coroutines log often and you just need to correlate log records
Roman Elizarov2f6d7c92017-02-03 15:16:07 +03001100coming from the same coroutine. However, when coroutine is tied to the processing of a specific request
1101or doing some specific background task, it is better to name it explicitly for debugging purposes.
Roman Elizarov66f018c2017-09-29 21:39:03 +03001102[CoroutineName] context element serves the same function as a thread name. It'll get displayed in the thread name that
1103is executing this coroutine when [debugging mode](#debugging-coroutines-and-threads) is turned on.
Roman Elizarov2f6d7c92017-02-03 15:16:07 +03001104
1105The following example demonstrates this concept:
1106
1107```kotlin
1108fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
1109
1110fun main(args: Array<String>) = runBlocking(CoroutineName("main")) {
1111 log("Started main coroutine")
1112 // run two background value computations
Roman Elizarov66f018c2017-09-29 21:39:03 +03001113 val v1 = async(CoroutineName("v1coroutine")) {
Roman Elizarov2f6d7c92017-02-03 15:16:07 +03001114 log("Computing v1")
1115 delay(500)
1116 252
1117 }
Roman Elizarov66f018c2017-09-29 21:39:03 +03001118 val v2 = async(CoroutineName("v2coroutine")) {
Roman Elizarov2f6d7c92017-02-03 15:16:07 +03001119 log("Computing v2")
1120 delay(1000)
1121 6
1122 }
1123 log("The answer for v1 / v2 = ${v1.await() / v2.await()}")
1124}
1125```
1126
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001127> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-context-09.kt)
Roman Elizarov2f6d7c92017-02-03 15:16:07 +03001128
1129The output it produces with `-Dkotlinx.coroutines.debug` JVM option is similar to:
1130
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001131```text
Roman Elizarov2f6d7c92017-02-03 15:16:07 +03001132[main @main#1] Started main coroutine
1133[ForkJoinPool.commonPool-worker-1 @v1coroutine#2] Computing v1
1134[ForkJoinPool.commonPool-worker-2 @v2coroutine#3] Computing v2
1135[main @main#1] The answer for v1 / v2 = 42
1136```
Roman Elizarov1293ccd2017-02-01 18:49:54 +03001137
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001138<!--- TEST FLEXIBLE_THREAD -->
1139
Roman Elizarov2fd7cb32017-02-11 23:18:59 +03001140### Cancellation via explicit job
1141
1142Let us put our knowledge about contexts, children and jobs together. Assume that our application has
1143an object with a lifecycle, but that object is not a coroutine. For example, we are writing an Android application
1144and launch various coroutines in the context of an Android activity to perform asynchronous operations to fetch
1145and update data, do animations, etc. All of these coroutines must be cancelled when activity is destroyed
1146to avoid memory leaks.
1147
1148We can manage a lifecycle of our coroutines by creating an instance of [Job] that is tied to
Roman Elizarov88396732017-09-27 21:30:47 +03001149the lifecycle of our activity. A job instance is created using [Job()] factory function
Roman Elizarov2fd7cb32017-02-11 23:18:59 +03001150as the following example shows. We need to make sure that all the coroutines are started
1151with this job in their context and then a single invocation of [Job.cancel] terminates them all.
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001152Moreover, [Job.join] waits for all of them to complete, so we can also use [cancelAndJoin] here in
1153this example:
Roman Elizarov2fd7cb32017-02-11 23:18:59 +03001154
1155```kotlin
1156fun main(args: Array<String>) = runBlocking<Unit> {
1157 val job = Job() // create a job object to manage our lifecycle
1158 // now launch ten coroutines for a demo, each working for a different time
1159 val coroutines = List(10) { i ->
1160 // they are all children of our job object
Roman Elizarov43e3af72017-07-21 16:01:31 +03001161 launch(coroutineContext + job) { // we use the context of main runBlocking thread, but with our own job object
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001162 delay((i + 1) * 200L) // variable delay 200ms, 400ms, ... etc
Roman Elizarov2fd7cb32017-02-11 23:18:59 +03001163 println("Coroutine $i is done")
1164 }
1165 }
1166 println("Launched ${coroutines.size} coroutines")
1167 delay(500L) // delay for half a second
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001168 println("Cancelling the job!")
1169 job.cancelAndJoin() // cancel all our coroutines and wait for all of them to complete
Roman Elizarov2fd7cb32017-02-11 23:18:59 +03001170}
1171```
1172
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001173> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-context-10.kt)
Roman Elizarov2fd7cb32017-02-11 23:18:59 +03001174
1175The output of this example is:
1176
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001177```text
Roman Elizarov2fd7cb32017-02-11 23:18:59 +03001178Launched 10 coroutines
1179Coroutine 0 is done
1180Coroutine 1 is done
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001181Cancelling the job!
Roman Elizarov2fd7cb32017-02-11 23:18:59 +03001182```
1183
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001184<!--- TEST -->
1185
Roman Elizarov2fd7cb32017-02-11 23:18:59 +03001186As you can see, only the first three coroutines had printed a message and the others were cancelled
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001187by a single invocation of `job.cancelAndJoin()`. So all we need to do in our hypothetical Android
Roman Elizarov2fd7cb32017-02-11 23:18:59 +03001188application is to create a parent job object when activity is created, use it for child coroutines,
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001189and cancel it when activity is destroyed. We cannot `join` them in the case of Android lifecycle,
1190since it is synchronous, but this joining ability is useful when building backend services to ensure bounded
1191resource usage.
Roman Elizarov2fd7cb32017-02-11 23:18:59 +03001192
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001193## Channels
Roman Elizarov7deefb82017-01-31 10:33:17 +03001194
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001195Deferred values provide a convenient way to transfer a single value between coroutines.
1196Channels provide a way to transfer a stream of values.
1197
1198<!--- INCLUDE .*/example-channel-([0-9]+).kt
1199import kotlinx.coroutines.experimental.channels.*
1200-->
1201
1202### Channel basics
1203
Roman Elizarov419a6c82017-02-09 18:36:22 +03001204A [Channel] is conceptually very similar to `BlockingQueue`. One key difference is that
1205instead of a blocking `put` operation it has a suspending [send][SendChannel.send], and instead of
1206a blocking `take` operation it has a suspending [receive][ReceiveChannel.receive].
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001207
1208```kotlin
1209fun main(args: Array<String>) = runBlocking<Unit> {
1210 val channel = Channel<Int>()
Roman Elizarov66f018c2017-09-29 21:39:03 +03001211 launch {
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001212 // this might be heavy CPU-consuming computation or async logic, we'll just send five squares
1213 for (x in 1..5) channel.send(x * x)
1214 }
1215 // here we print five received integers:
1216 repeat(5) { println(channel.receive()) }
1217 println("Done!")
1218}
1219```
1220
Roman Elizarove8d79342017-08-29 15:21:21 +03001221> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-channel-01.kt)
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001222
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001223The output of this code is:
1224
1225```text
12261
12274
12289
122916
123025
1231Done!
1232```
1233
1234<!--- TEST -->
1235
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001236### Closing and iteration over channels
1237
1238Unlike a queue, a channel can be closed to indicate that no more elements are coming.
1239On the receiver side it is convenient to use a regular `for` loop to receive elements
1240from the channel.
1241
Roman Elizarov419a6c82017-02-09 18:36:22 +03001242Conceptually, a [close][SendChannel.close] is like sending a special close token to the channel.
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001243The iteration stops as soon as this close token is received, so there is a guarantee
1244that all previously sent elements before the close are received:
1245
1246```kotlin
1247fun main(args: Array<String>) = runBlocking<Unit> {
1248 val channel = Channel<Int>()
Roman Elizarov66f018c2017-09-29 21:39:03 +03001249 launch {
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001250 for (x in 1..5) channel.send(x * x)
1251 channel.close() // we're done sending
1252 }
1253 // here we print received values using `for` loop (until the channel is closed)
1254 for (y in channel) println(y)
1255 println("Done!")
1256}
1257```
1258
Roman Elizarove8d79342017-08-29 15:21:21 +03001259> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-channel-02.kt)
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001260
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001261<!--- TEST
12621
12634
12649
126516
126625
1267Done!
1268-->
1269
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001270### Building channel producers
1271
Roman Elizarova5e653f2017-02-13 13:49:55 +03001272The pattern where a coroutine is producing a sequence of elements is quite common.
1273This is a part of _producer-consumer_ pattern that is often found in concurrent code.
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001274You could abstract such a producer into a function that takes channel as its parameter, but this goes contrary
Roman Elizarova5e653f2017-02-13 13:49:55 +03001275to common sense that results must be returned from functions.
1276
Roman Elizarov86349be2017-03-17 16:47:37 +03001277There is a convenience coroutine builder named [produce] that makes it easy to do it right on producer side,
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001278and an extension function [consumeEach], that replaces a `for` loop on the consumer side:
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001279
1280```kotlin
Roman Elizarov66f018c2017-09-29 21:39:03 +03001281fun produceSquares() = produce<Int> {
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001282 for (x in 1..5) send(x * x)
1283}
1284
1285fun main(args: Array<String>) = runBlocking<Unit> {
1286 val squares = produceSquares()
Roman Elizarov86349be2017-03-17 16:47:37 +03001287 squares.consumeEach { println(it) }
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001288 println("Done!")
1289}
1290```
1291
Roman Elizarove8d79342017-08-29 15:21:21 +03001292> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-channel-03.kt)
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001293
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001294<!--- TEST
12951
12964
12979
129816
129925
1300Done!
1301-->
1302
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001303### Pipelines
1304
Roman Elizarov66f018c2017-09-29 21:39:03 +03001305A pipeline is a pattern where one coroutine is producing, possibly infinite, stream of values:
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001306
1307```kotlin
Roman Elizarov66f018c2017-09-29 21:39:03 +03001308fun produceNumbers() = produce<Int> {
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001309 var x = 1
1310 while (true) send(x++) // infinite stream of integers starting from 1
1311}
1312```
1313
Roman Elizarova5e653f2017-02-13 13:49:55 +03001314And another coroutine or coroutines are consuming that stream, doing some processing, and producing some other results.
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001315In the below example the numbers are just squared:
1316
1317```kotlin
Roman Elizarov66f018c2017-09-29 21:39:03 +03001318fun square(numbers: ReceiveChannel<Int>) = produce<Int> {
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001319 for (x in numbers) send(x * x)
1320}
1321```
1322
Roman Elizarova5e653f2017-02-13 13:49:55 +03001323The main code starts and connects the whole pipeline:
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001324
1325```kotlin
1326fun main(args: Array<String>) = runBlocking<Unit> {
1327 val numbers = produceNumbers() // produces integers from 1 and on
1328 val squares = square(numbers) // squares integers
1329 for (i in 1..5) println(squares.receive()) // print first five
1330 println("Done!") // we are done
1331 squares.cancel() // need to cancel these coroutines in a larger app
1332 numbers.cancel()
1333}
1334```
1335
Roman Elizarove8d79342017-08-29 15:21:21 +03001336> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-channel-04.kt)
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001337
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001338<!--- TEST
13391
13404
13419
134216
134325
1344Done!
1345-->
1346
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001347We don't have to cancel these coroutines in this example app, because
1348[coroutines are like daemon threads](#coroutines-are-like-daemon-threads),
1349but in a larger app we'll need to stop our pipeline if we don't need it anymore.
1350Alternatively, we could have run pipeline coroutines as
Roman Elizarov66f018c2017-09-29 21:39:03 +03001351[children of a main coroutine](#children-of-a-coroutine) as is demonstrated in the following example.
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001352
1353### Prime numbers with pipeline
1354
Cedric Beustfa0b28f2017-02-07 07:07:25 -08001355Let's take pipelines to the extreme with an example that generates prime numbers using a pipeline
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001356of coroutines. We start with an infinite sequence of numbers. This time we introduce an
Roman Elizarov66f018c2017-09-29 21:39:03 +03001357explicit `context` parameter and pass it to [produce] builder,
1358so that caller can control where our coroutines run:
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001359
Roman Elizarove8d79342017-08-29 15:21:21 +03001360<!--- INCLUDE core/kotlinx-coroutines-core/src/test/kotlin/guide/example-channel-05.kt
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001361import kotlin.coroutines.experimental.CoroutineContext
1362-->
1363
1364```kotlin
Roman Elizarova5e653f2017-02-13 13:49:55 +03001365fun numbersFrom(context: CoroutineContext, start: Int) = produce<Int>(context) {
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001366 var x = start
1367 while (true) send(x++) // infinite stream of integers from start
1368}
1369```
1370
1371The following pipeline stage filters an incoming stream of numbers, removing all the numbers
1372that are divisible by the given prime number:
1373
1374```kotlin
Roman Elizarova5e653f2017-02-13 13:49:55 +03001375fun filter(context: CoroutineContext, numbers: ReceiveChannel<Int>, prime: Int) = produce<Int>(context) {
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001376 for (x in numbers) if (x % prime != 0) send(x)
1377}
1378```
1379
1380Now we build our pipeline by starting a stream of numbers from 2, taking a prime number from the current channel,
Roman Elizarov62500ba2017-02-09 18:55:40 +03001381and launching new pipeline stage for each prime number found:
1382
1383```
Roman Elizarova5e653f2017-02-13 13:49:55 +03001384numbersFrom(2) -> filter(2) -> filter(3) -> filter(5) -> filter(7) ...
Roman Elizarov62500ba2017-02-09 18:55:40 +03001385```
1386
1387The following example prints the first ten prime numbers,
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001388running the whole pipeline in the context of the main thread. Since all the coroutines are launched as
1389children of the main [runBlocking] coroutine in its [coroutineContext][CoroutineScope.coroutineContext],
Roman Elizarov66f018c2017-09-29 21:39:03 +03001390we don't have to keep an explicit list of all the coroutine we have started.
Roman Elizarov88396732017-09-27 21:30:47 +03001391We use [cancelChildren] extension function to cancel all the children coroutines.
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001392
1393```kotlin
1394fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov43e3af72017-07-21 16:01:31 +03001395 var cur = numbersFrom(coroutineContext, 2)
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001396 for (i in 1..10) {
1397 val prime = cur.receive()
1398 println(prime)
Roman Elizarov43e3af72017-07-21 16:01:31 +03001399 cur = filter(coroutineContext, cur, prime)
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001400 }
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001401 coroutineContext.cancelChildren() // cancel all children to let main finish
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001402}
1403```
1404
Roman Elizarove8d79342017-08-29 15:21:21 +03001405> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-channel-05.kt)
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001406
1407The output of this code is:
1408
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001409```text
Roman Elizarovb7721cf2017-02-03 19:23:08 +030014102
14113
14125
14137
141411
141513
141617
141719
141823
141929
1420```
1421
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001422<!--- TEST -->
1423
Roman Elizarov66f018c2017-09-29 21:39:03 +03001424Note, that you can build the same pipeline using
1425[`buildIterator`](https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.coroutines.experimental/build-iterator.html)
1426coroutine builder from the standard library.
Roman Elizarova5e653f2017-02-13 13:49:55 +03001427Replace `produce` with `buildIterator`, `send` with `yield`, `receive` with `next`,
Roman Elizarov62500ba2017-02-09 18:55:40 +03001428`ReceiveChannel` with `Iterator`, and get rid of the context. You will not need `runBlocking` either.
1429However, the benefit of a pipeline that uses channels as shown above is that it can actually use
1430multiple CPU cores if you run it in [CommonPool] context.
1431
Roman Elizarova5e653f2017-02-13 13:49:55 +03001432Anyway, this is an extremely impractical way to find prime numbers. In practice, pipelines do involve some
Roman Elizarov62500ba2017-02-09 18:55:40 +03001433other suspending invocations (like asynchronous calls to remote services) and these pipelines cannot be
1434built using `buildSeqeunce`/`buildIterator`, because they do not allow arbitrary suspension, unlike
Roman Elizarov66f018c2017-09-29 21:39:03 +03001435`produce`, which is fully asynchronous.
Roman Elizarov62500ba2017-02-09 18:55:40 +03001436
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001437### Fan-out
1438
1439Multiple coroutines may receive from the same channel, distributing work between themselves.
1440Let us start with a producer coroutine that is periodically producing integers
1441(ten numbers per second):
1442
1443```kotlin
Roman Elizarov66f018c2017-09-29 21:39:03 +03001444fun produceNumbers() = produce<Int> {
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001445 var x = 1 // start from 1
1446 while (true) {
1447 send(x++) // produce next
1448 delay(100) // wait 0.1s
1449 }
1450}
1451```
1452
1453Then we can have several processor coroutines. In this example, they just print their id and
1454received number:
1455
1456```kotlin
Roman Elizarov66f018c2017-09-29 21:39:03 +03001457fun launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch {
Roman Elizarov86349be2017-03-17 16:47:37 +03001458 channel.consumeEach {
1459 println("Processor #$id received $it")
Roman Elizarovec9384c2017-03-02 22:09:08 +03001460 }
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001461}
1462```
1463
Roman Elizarov35d2c342017-07-20 14:54:39 +03001464Now let us launch five processors and let them work for almost a second. See what happens:
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001465
1466```kotlin
1467fun main(args: Array<String>) = runBlocking<Unit> {
1468 val producer = produceNumbers()
1469 repeat(5) { launchProcessor(it, producer) }
Roman Elizarov35d2c342017-07-20 14:54:39 +03001470 delay(950)
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001471 producer.cancel() // cancel producer coroutine and thus kill them all
1472}
1473```
1474
Roman Elizarove8d79342017-08-29 15:21:21 +03001475> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-channel-06.kt)
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001476
1477The output will be similar to the the following one, albeit the processor ids that receive
1478each specific integer may be different:
1479
1480```
1481Processor #2 received 1
1482Processor #4 received 2
1483Processor #0 received 3
1484Processor #1 received 4
1485Processor #3 received 5
1486Processor #2 received 6
1487Processor #4 received 7
1488Processor #0 received 8
1489Processor #1 received 9
1490Processor #3 received 10
1491```
1492
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001493<!--- TEST lines.size == 10 && lines.withIndex().all { (i, line) -> line.startsWith("Processor #") && line.endsWith(" received ${i + 1}") } -->
1494
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001495Note, that cancelling a producer coroutine closes its channel, thus eventually terminating iteration
1496over the channel that processor coroutines are doing.
1497
1498### Fan-in
1499
1500Multiple coroutines may send to the same channel.
1501For example, let us have a channel of strings, and a suspending function that
1502repeatedly sends a specified string to this channel with a specified delay:
1503
1504```kotlin
1505suspend fun sendString(channel: SendChannel<String>, s: String, time: Long) {
1506 while (true) {
1507 delay(time)
1508 channel.send(s)
1509 }
1510}
1511```
1512
Cedric Beustfa0b28f2017-02-07 07:07:25 -08001513Now, let us see what happens if we launch a couple of coroutines sending strings
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001514(in this example we launch them in the context of the main thread as main coroutine's children):
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001515
1516```kotlin
1517fun main(args: Array<String>) = runBlocking<Unit> {
1518 val channel = Channel<String>()
Roman Elizarov43e3af72017-07-21 16:01:31 +03001519 launch(coroutineContext) { sendString(channel, "foo", 200L) }
1520 launch(coroutineContext) { sendString(channel, "BAR!", 500L) }
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001521 repeat(6) { // receive first six
1522 println(channel.receive())
1523 }
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001524 coroutineContext.cancelChildren() // cancel all children to let main finish
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001525}
1526```
1527
Roman Elizarove8d79342017-08-29 15:21:21 +03001528> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-channel-07.kt)
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001529
1530The output is:
1531
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001532```text
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001533foo
1534foo
1535BAR!
1536foo
1537foo
1538BAR!
1539```
1540
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001541<!--- TEST -->
1542
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001543### Buffered channels
1544
1545The channels shown so far had no buffer. Unbuffered channels transfer elements when sender and receiver
1546meet each other (aka rendezvous). If send is invoked first, then it is suspended until receive is invoked,
1547if receive is invoked first, it is suspended until send is invoked.
Roman Elizarov419a6c82017-02-09 18:36:22 +03001548
Roman Elizarov88396732017-09-27 21:30:47 +03001549Both [Channel()] factory function and [produce] builder take an optional `capacity` parameter to
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001550specify _buffer size_. Buffer allows senders to send multiple elements before suspending,
1551similar to the `BlockingQueue` with a specified capacity, which blocks when buffer is full.
1552
1553Take a look at the behavior of the following code:
1554
1555```kotlin
1556fun main(args: Array<String>) = runBlocking<Unit> {
1557 val channel = Channel<Int>(4) // create buffered channel
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001558 val sender = launch(coroutineContext) { // launch sender coroutine
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001559 repeat(10) {
1560 println("Sending $it") // print before sending each element
1561 channel.send(it) // will suspend when buffer is full
1562 }
1563 }
1564 // don't receive anything... just wait....
1565 delay(1000)
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001566 sender.cancel() // cancel sender coroutine
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001567}
1568```
1569
Roman Elizarove8d79342017-08-29 15:21:21 +03001570> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-channel-08.kt)
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001571
1572It prints "sending" _five_ times using a buffered channel with capacity of _four_:
1573
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001574```text
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001575Sending 0
1576Sending 1
1577Sending 2
1578Sending 3
1579Sending 4
1580```
1581
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001582<!--- TEST -->
1583
Roman Elizarovb7721cf2017-02-03 19:23:08 +03001584The first four elements are added to the buffer and the sender suspends when trying to send the fifth one.
Roman Elizarov419a6c82017-02-09 18:36:22 +03001585
Roman Elizarovb0517ba2017-02-27 14:03:14 +03001586### Channels are fair
1587
1588Send and receive operations to channels are _fair_ with respect to the order of their invocation from
1589multiple coroutines. They are served in first-in first-out order, e.g. the first coroutine to invoke `receive`
1590gets the element. In the following example two coroutines "ping" and "pong" are
1591receiving the "ball" object from the shared "table" channel.
1592
1593```kotlin
1594data class Ball(var hits: Int)
1595
1596fun main(args: Array<String>) = runBlocking<Unit> {
1597 val table = Channel<Ball>() // a shared table
Roman Elizarov43e3af72017-07-21 16:01:31 +03001598 launch(coroutineContext) { player("ping", table) }
1599 launch(coroutineContext) { player("pong", table) }
Roman Elizarovb0517ba2017-02-27 14:03:14 +03001600 table.send(Ball(0)) // serve the ball
1601 delay(1000) // delay 1 second
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001602 coroutineContext.cancelChildren() // game over, cancel them
Roman Elizarovb0517ba2017-02-27 14:03:14 +03001603}
1604
1605suspend fun player(name: String, table: Channel<Ball>) {
1606 for (ball in table) { // receive the ball in a loop
1607 ball.hits++
1608 println("$name $ball")
Roman Elizarovf526b132017-03-10 16:07:14 +03001609 delay(300) // wait a bit
Roman Elizarovb0517ba2017-02-27 14:03:14 +03001610 table.send(ball) // send the ball back
1611 }
1612}
1613```
1614
Roman Elizarove8d79342017-08-29 15:21:21 +03001615> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-channel-09.kt)
Roman Elizarovb0517ba2017-02-27 14:03:14 +03001616
1617The "ping" coroutine is started first, so it is the first one to receive the ball. Even though "ping"
1618coroutine immediately starts receiving the ball again after sending it back to the table, the ball gets
1619received by the "pong" coroutine, because it was already waiting for it:
1620
1621```text
1622ping Ball(hits=1)
1623pong Ball(hits=2)
1624ping Ball(hits=3)
1625pong Ball(hits=4)
Roman Elizarovb0517ba2017-02-27 14:03:14 +03001626```
1627
1628<!--- TEST -->
1629
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001630Note, that sometimes channels may produce executions that look unfair due to the nature of the executor
1631that is being used. See [this issue](https://github.com/Kotlin/kotlinx.coroutines/issues/111) for details.
1632
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001633## Shared mutable state and concurrency
1634
Roman Elizarov66f018c2017-09-29 21:39:03 +03001635Coroutines can be executed concurrently using a multi-threaded dispatcher like the default [CommonPool]. It presents
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001636all the usual concurrency problems. The main problem being synchronization of access to **shared mutable state**.
1637Some solutions to this problem in the land of coroutines are similar to the solutions in the multi-threaded world,
1638but others are unique.
1639
1640### The problem
1641
Roman Elizarov1e459602017-02-27 11:05:17 +03001642Let us launch a thousand coroutines all doing the same action thousand times (for a total of a million executions).
1643We'll also measure their completion time for further comparisons:
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001644
Roman Elizarov43e90112017-05-10 11:25:20 +03001645<!--- INCLUDE .*/example-sync-([0-9a-z]+).kt
Roman Elizarov1e459602017-02-27 11:05:17 +03001646import kotlin.coroutines.experimental.CoroutineContext
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001647import kotlin.system.measureTimeMillis
1648-->
1649
Roman Elizarov1e459602017-02-27 11:05:17 +03001650<!--- INCLUDE .*/example-sync-03.kt
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001651import java.util.concurrent.atomic.AtomicInteger
1652-->
1653
Roman Elizarov1e459602017-02-27 11:05:17 +03001654<!--- INCLUDE .*/example-sync-06.kt
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001655import kotlinx.coroutines.experimental.sync.Mutex
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001656import kotlinx.coroutines.experimental.sync.withLock
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001657-->
1658
Roman Elizarov1e459602017-02-27 11:05:17 +03001659<!--- INCLUDE .*/example-sync-07.kt
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001660import kotlinx.coroutines.experimental.channels.*
1661-->
1662
1663```kotlin
Roman Elizarov1e459602017-02-27 11:05:17 +03001664suspend fun massiveRun(context: CoroutineContext, action: suspend () -> Unit) {
1665 val n = 1000 // number of coroutines to launch
1666 val k = 1000 // times an action is repeated by each coroutine
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001667 val time = measureTimeMillis {
1668 val jobs = List(n) {
Roman Elizarov1e459602017-02-27 11:05:17 +03001669 launch(context) {
1670 repeat(k) { action() }
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001671 }
1672 }
1673 jobs.forEach { it.join() }
1674 }
Roman Elizarov1e459602017-02-27 11:05:17 +03001675 println("Completed ${n * k} actions in $time ms")
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001676}
1677```
1678
Roman Elizarov43e90112017-05-10 11:25:20 +03001679<!--- INCLUDE .*/example-sync-([0-9a-z]+).kt -->
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001680
Roman Elizarov1e459602017-02-27 11:05:17 +03001681We start with a very simple action that increments a shared mutable variable using
1682multi-threaded [CommonPool] context.
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001683
1684```kotlin
1685var counter = 0
1686
1687fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov1e459602017-02-27 11:05:17 +03001688 massiveRun(CommonPool) {
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001689 counter++
1690 }
1691 println("Counter = $counter")
1692}
1693```
1694
Roman Elizarove8d79342017-08-29 15:21:21 +03001695> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-01.kt)
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001696
Roman Elizarov1e459602017-02-27 11:05:17 +03001697<!--- TEST LINES_START
1698Completed 1000000 actions in
1699Counter =
1700-->
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001701
Roman Elizarov1e459602017-02-27 11:05:17 +03001702What does it print at the end? It is highly unlikely to ever print "Counter = 1000000", because a thousand coroutines
1703increment the `counter` concurrently from multiple threads without any synchronization.
1704
Roman Elizarov43e90112017-05-10 11:25:20 +03001705> Note: if you have an old system with 2 or fewer CPUs, then you _will_ consistently see 1000000, because
1706`CommonPool` is running in only one thread in this case. To reproduce the problem you'll need to make the
1707following change:
1708
1709```kotlin
1710val mtContext = newFixedThreadPoolContext(2, "mtPool") // explicitly define context with two threads
1711var counter = 0
1712
1713fun main(args: Array<String>) = runBlocking<Unit> {
1714 massiveRun(mtContext) { // use it instead of CommonPool in this sample and below
1715 counter++
1716 }
1717 println("Counter = $counter")
1718}
1719```
1720
Roman Elizarove8d79342017-08-29 15:21:21 +03001721> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-01b.kt)
Roman Elizarov43e90112017-05-10 11:25:20 +03001722
1723<!--- TEST LINES_START
1724Completed 1000000 actions in
1725Counter =
1726-->
1727
Roman Elizarov1e459602017-02-27 11:05:17 +03001728### Volatiles are of no help
1729
1730There is common misconception that making a variable `volatile` solves concurrency problem. Let us try it:
1731
1732```kotlin
1733@Volatile // in Kotlin `volatile` is an annotation
1734var counter = 0
1735
1736fun main(args: Array<String>) = runBlocking<Unit> {
1737 massiveRun(CommonPool) {
1738 counter++
1739 }
1740 println("Counter = $counter")
1741}
1742```
1743
Roman Elizarove8d79342017-08-29 15:21:21 +03001744> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-02.kt)
Roman Elizarov1e459602017-02-27 11:05:17 +03001745
1746<!--- TEST LINES_START
1747Completed 1000000 actions in
1748Counter =
1749-->
1750
1751This code works slower, but we still don't get "Counter = 1000000" at the end, because volatile variables guarantee
1752linearizable (this is a technical term for "atomic") reads and writes to the corresponding variable, but
1753do not provide atomicity of larger actions (increment in our case).
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001754
1755### Thread-safe data structures
1756
1757The general solution that works both for threads and for coroutines is to use a thread-safe (aka synchronized,
1758linearizable, or atomic) data structure that provides all the necessarily synchronization for the corresponding
1759operations that needs to be performed on a shared state.
Roman Elizarov1e459602017-02-27 11:05:17 +03001760In the case of a simple counter we can use `AtomicInteger` class which has atomic `incrementAndGet` operations:
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001761
1762```kotlin
1763var counter = AtomicInteger()
1764
1765fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov1e459602017-02-27 11:05:17 +03001766 massiveRun(CommonPool) {
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001767 counter.incrementAndGet()
1768 }
1769 println("Counter = ${counter.get()}")
1770}
1771```
1772
Roman Elizarove8d79342017-08-29 15:21:21 +03001773> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-03.kt)
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001774
Roman Elizarov1e459602017-02-27 11:05:17 +03001775<!--- TEST ARBITRARY_TIME
1776Completed 1000000 actions in xxx ms
1777Counter = 1000000
1778-->
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001779
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001780This is the fastest solution for this particular problem. It works for plain counters, collections, queues and other
1781standard data structures and basic operations on them. However, it does not easily scale to complex
1782state or to complex operations that do not have ready-to-use thread-safe implementations.
1783
Roman Elizarov1e459602017-02-27 11:05:17 +03001784### Thread confinement fine-grained
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001785
Roman Elizarov1e459602017-02-27 11:05:17 +03001786_Thread confinement_ is an approach to the problem of shared mutable state where all access to the particular shared
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001787state is confined to a single thread. It is typically used in UI applications, where all UI state is confined to
1788the single event-dispatch/application thread. It is easy to apply with coroutines by using a
1789single-threaded context:
1790
1791```kotlin
1792val counterContext = newSingleThreadContext("CounterContext")
1793var counter = 0
1794
1795fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov1e459602017-02-27 11:05:17 +03001796 massiveRun(CommonPool) { // run each coroutine in CommonPool
1797 run(counterContext) { // but confine each increment to the single-threaded context
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001798 counter++
1799 }
1800 }
1801 println("Counter = $counter")
1802}
1803```
1804
Roman Elizarove8d79342017-08-29 15:21:21 +03001805> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-04.kt)
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001806
Roman Elizarov1e459602017-02-27 11:05:17 +03001807<!--- TEST ARBITRARY_TIME
1808Completed 1000000 actions in xxx ms
1809Counter = 1000000
1810-->
1811
1812This code works very slowly, because it does _fine-grained_ thread-confinement. Each individual increment switches
1813from multi-threaded `CommonPool` context to the single-threaded context using [run] block.
1814
1815### Thread confinement coarse-grained
1816
1817In practice, thread confinement is performed in large chunks, e.g. big pieces of state-updating business logic
1818are confined to the single thread. The following example does it like that, running each coroutine in
1819the single-threaded context to start with.
1820
1821```kotlin
1822val counterContext = newSingleThreadContext("CounterContext")
1823var counter = 0
1824
1825fun main(args: Array<String>) = runBlocking<Unit> {
1826 massiveRun(counterContext) { // run each coroutine in the single-threaded context
1827 counter++
1828 }
1829 println("Counter = $counter")
1830}
1831```
1832
Roman Elizarove8d79342017-08-29 15:21:21 +03001833> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-05.kt)
Roman Elizarov1e459602017-02-27 11:05:17 +03001834
1835<!--- TEST ARBITRARY_TIME
1836Completed 1000000 actions in xxx ms
1837Counter = 1000000
1838-->
1839
1840This now works much faster and produces correct result.
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001841
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001842### Mutual exclusion
1843
1844Mutual exclusion solution to the problem is to protect all modifications of the shared state with a _critical section_
1845that is never executed concurrently. In a blocking world you'd typically use `synchronized` or `ReentrantLock` for that.
1846Coroutine's alternative is called [Mutex]. It has [lock][Mutex.lock] and [unlock][Mutex.unlock] functions to
1847delimit a critical section. The key difference is that `Mutex.lock` is a suspending function. It does not block a thread.
1848
Roman Elizarov88396732017-09-27 21:30:47 +03001849There is also [withLock] extension function that conveniently represents
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001850`mutex.lock(); try { ... } finally { mutex.unlock() }` pattern:
1851
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001852```kotlin
1853val mutex = Mutex()
1854var counter = 0
1855
1856fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov1e459602017-02-27 11:05:17 +03001857 massiveRun(CommonPool) {
Roman Elizarov8b38fa22017-09-27 17:44:31 +03001858 mutex.withLock {
1859 counter++
1860 }
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001861 }
1862 println("Counter = $counter")
1863}
1864```
1865
Roman Elizarove8d79342017-08-29 15:21:21 +03001866> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-06.kt)
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001867
Roman Elizarov1e459602017-02-27 11:05:17 +03001868<!--- TEST ARBITRARY_TIME
1869Completed 1000000 actions in xxx ms
1870Counter = 1000000
1871-->
1872
1873The locking in this example is fine-grained, so it pays the price. However, it is a good choice for some situations
1874where you absolutely must modify some shared state periodically, but there is no natural thread that this state
1875is confined to.
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001876
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001877### Actors
1878
1879An actor is a combination of a coroutine, the state that is confined and is encapsulated into this coroutine,
1880and a channel to communicate with other coroutines. A simple actor can be written as a function,
1881but an actor with a complex state is better suited for a class.
1882
Roman Elizarovc0e19f82017-02-27 11:59:14 +03001883There is an [actor] coroutine builder that conveniently combines actor's mailbox channel into its
1884scope to receive messages from and combines the send channel into the resulting job object, so that a
1885single reference to the actor can be carried around as its handle.
1886
Roman Elizarov256812a2017-07-22 01:00:30 +03001887The first step of using an actor is to define a class of messages that an actor is going to process.
1888Kotlin's [sealed classes](https://kotlinlang.org/docs/reference/sealed-classes.html) are well suited for that purpose.
1889We define `CounterMsg` sealed class with `IncCounter` message to increment a counter and `GetCounter` message
1890to get its value. The later needs to send a response. A [CompletableDeferred] communication
1891primitive, that represents a single value that will be known (communicated) in the future,
1892is used here for that purpose.
1893
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001894```kotlin
1895// Message types for counterActor
1896sealed class CounterMsg
1897object IncCounter : CounterMsg() // one-way message to increment counter
Roman Elizarov256812a2017-07-22 01:00:30 +03001898class GetCounter(val response: CompletableDeferred<Int>) : CounterMsg() // a request with reply
1899```
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001900
Roman Elizarov256812a2017-07-22 01:00:30 +03001901Then we define a function that launches an actor using an [actor] coroutine builder:
1902
1903```kotlin
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001904// This function launches a new counter actor
Roman Elizarov66f018c2017-09-29 21:39:03 +03001905fun counterActor() = actor<CounterMsg> {
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001906 var counter = 0 // actor state
Roman Elizarovc0e19f82017-02-27 11:59:14 +03001907 for (msg in channel) { // iterate over incoming messages
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001908 when (msg) {
1909 is IncCounter -> counter++
Roman Elizarov256812a2017-07-22 01:00:30 +03001910 is GetCounter -> msg.response.complete(counter)
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001911 }
1912 }
1913}
Roman Elizarov256812a2017-07-22 01:00:30 +03001914```
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001915
Roman Elizarov256812a2017-07-22 01:00:30 +03001916The main code is straightforward:
1917
1918```kotlin
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001919fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarovc0e19f82017-02-27 11:59:14 +03001920 val counter = counterActor() // create the actor
Roman Elizarov1e459602017-02-27 11:05:17 +03001921 massiveRun(CommonPool) {
Roman Elizarovc0e19f82017-02-27 11:59:14 +03001922 counter.send(IncCounter)
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001923 }
Roman Elizarov256812a2017-07-22 01:00:30 +03001924 // send a message to get a counter value from an actor
1925 val response = CompletableDeferred<Int>()
Roman Elizarovc0e19f82017-02-27 11:59:14 +03001926 counter.send(GetCounter(response))
Roman Elizarov256812a2017-07-22 01:00:30 +03001927 println("Counter = ${response.await()}")
Roman Elizarovc0e19f82017-02-27 11:59:14 +03001928 counter.close() // shutdown the actor
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001929}
1930```
1931
Roman Elizarove8d79342017-08-29 15:21:21 +03001932> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-07.kt)
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001933
Roman Elizarov1e459602017-02-27 11:05:17 +03001934<!--- TEST ARBITRARY_TIME
1935Completed 1000000 actions in xxx ms
1936Counter = 1000000
1937-->
Roman Elizarov731f0ad2017-02-22 20:48:45 +03001938
Roman Elizarovc0e19f82017-02-27 11:59:14 +03001939It does not matter (for correctness) what context the actor itself is executed in. An actor is
Roman Elizarovf5bc0472017-02-22 11:38:13 +03001940a coroutine and a coroutine is executed sequentially, so confinement of the state to the specific coroutine
1941works as a solution to the problem of shared mutable state.
1942
Roman Elizarovc0e19f82017-02-27 11:59:14 +03001943Actor is more efficient than locking under load, because in this case it always has work to do and it does not
1944have to switch to a different context at all.
1945
1946> Note, that an [actor] coroutine builder is a dual of [produce] coroutine builder. An actor is associated
1947 with the channel that it receives messages from, while a producer is associated with the channel that it
1948 sends elements to.
Roman Elizarov1e459602017-02-27 11:05:17 +03001949
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03001950## Select expression
1951
Roman Elizarova84730b2017-02-22 11:58:50 +03001952Select expression makes it possible to await multiple suspending functions simultaneously and _select_
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03001953the first one that becomes available.
1954
1955<!--- INCLUDE .*/example-select-([0-9]+).kt
1956import kotlinx.coroutines.experimental.channels.*
1957import kotlinx.coroutines.experimental.selects.*
1958-->
1959
1960### Selecting from channels
1961
Roman Elizarov57857202017-03-02 23:17:25 +03001962Let us have two producers of strings: `fizz` and `buzz`. The `fizz` produces "Fizz" string every 300 ms:
1963
1964<!--- INCLUDE .*/example-select-01.kt
1965import kotlin.coroutines.experimental.CoroutineContext
1966-->
1967
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03001968```kotlin
Roman Elizarov57857202017-03-02 23:17:25 +03001969fun fizz(context: CoroutineContext) = produce<String>(context) {
1970 while (true) { // sends "Fizz" every 300 ms
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03001971 delay(300)
1972 send("Fizz")
1973 }
1974}
1975```
1976
Roman Elizarov57857202017-03-02 23:17:25 +03001977And the `buzz` produces "Buzz!" string every 500 ms:
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03001978
1979```kotlin
Roman Elizarov57857202017-03-02 23:17:25 +03001980fun buzz(context: CoroutineContext) = produce<String>(context) {
1981 while (true) { // sends "Buzz!" every 500 ms
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03001982 delay(500)
1983 send("Buzz!")
1984 }
1985}
1986```
1987
1988Using [receive][ReceiveChannel.receive] suspending function we can receive _either_ from one channel or the
1989other. But [select] expression allows us to receive from _both_ simultaneously using its
Roman Elizarov8a5564d2017-09-06 18:48:22 +03001990[onReceive][ReceiveChannel.onReceive] clauses:
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03001991
1992```kotlin
Roman Elizarov57857202017-03-02 23:17:25 +03001993suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) {
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03001994 select<Unit> { // <Unit> means that this select expression does not produce any result
1995 fizz.onReceive { value -> // this is the first select clause
1996 println("fizz -> '$value'")
1997 }
1998 buzz.onReceive { value -> // this is the second select clause
1999 println("buzz -> '$value'")
2000 }
2001 }
2002}
2003```
2004
Roman Elizarov57857202017-03-02 23:17:25 +03002005Let us run it all seven times:
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002006
2007```kotlin
2008fun main(args: Array<String>) = runBlocking<Unit> {
Roman Elizarov43e3af72017-07-21 16:01:31 +03002009 val fizz = fizz(coroutineContext)
2010 val buzz = buzz(coroutineContext)
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002011 repeat(7) {
Roman Elizarov57857202017-03-02 23:17:25 +03002012 selectFizzBuzz(fizz, buzz)
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002013 }
Roman Elizarov8b38fa22017-09-27 17:44:31 +03002014 coroutineContext.cancelChildren() // cancel fizz & buzz coroutines
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002015}
2016```
2017
Roman Elizarove8d79342017-08-29 15:21:21 +03002018> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-select-01.kt)
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002019
2020The result of this code is:
2021
Roman Elizarov731f0ad2017-02-22 20:48:45 +03002022```text
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002023fizz -> 'Fizz'
2024buzz -> 'Buzz!'
2025fizz -> 'Fizz'
2026fizz -> 'Fizz'
2027buzz -> 'Buzz!'
2028fizz -> 'Fizz'
2029buzz -> 'Buzz!'
2030```
2031
Roman Elizarov731f0ad2017-02-22 20:48:45 +03002032<!--- TEST -->
2033
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002034### Selecting on close
2035
Roman Elizarov8a5564d2017-09-06 18:48:22 +03002036The [onReceive][ReceiveChannel.onReceive] clause in `select` fails when the channel is closed and the corresponding
2037`select` throws an exception. We can use [onReceiveOrNull][ReceiveChannel.onReceiveOrNull] clause to perform a
Roman Elizarova84730b2017-02-22 11:58:50 +03002038specific action when the channel is closed. The following example also shows that `select` is an expression that returns
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002039the result of its selected clause:
2040
2041```kotlin
2042suspend fun selectAorB(a: ReceiveChannel<String>, b: ReceiveChannel<String>): String =
2043 select<String> {
2044 a.onReceiveOrNull { value ->
2045 if (value == null)
2046 "Channel 'a' is closed"
2047 else
2048 "a -> '$value'"
2049 }
2050 b.onReceiveOrNull { value ->
2051 if (value == null)
2052 "Channel 'b' is closed"
2053 else
2054 "b -> '$value'"
2055 }
2056 }
2057```
2058
Roman Elizarova84730b2017-02-22 11:58:50 +03002059Let's use it with channel `a` that produces "Hello" string four times and
2060channel `b` that produces "World" four times:
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002061
2062```kotlin
2063fun main(args: Array<String>) = runBlocking<Unit> {
2064 // we are using the context of the main thread in this example for predictability ...
Roman Elizarov43e3af72017-07-21 16:01:31 +03002065 val a = produce<String>(coroutineContext) {
Roman Elizarova84730b2017-02-22 11:58:50 +03002066 repeat(4) { send("Hello $it") }
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002067 }
Roman Elizarov43e3af72017-07-21 16:01:31 +03002068 val b = produce<String>(coroutineContext) {
Roman Elizarova84730b2017-02-22 11:58:50 +03002069 repeat(4) { send("World $it") }
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002070 }
2071 repeat(8) { // print first eight results
2072 println(selectAorB(a, b))
2073 }
Roman Elizarov8b38fa22017-09-27 17:44:31 +03002074 coroutineContext.cancelChildren()
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002075}
2076```
2077
Roman Elizarove8d79342017-08-29 15:21:21 +03002078> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-select-02.kt)
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002079
Roman Elizarova84730b2017-02-22 11:58:50 +03002080The result of this code is quite interesting, so we'll analyze it in mode detail:
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002081
Roman Elizarov731f0ad2017-02-22 20:48:45 +03002082```text
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002083a -> 'Hello 0'
2084a -> 'Hello 1'
2085b -> 'World 0'
2086a -> 'Hello 2'
2087a -> 'Hello 3'
2088b -> 'World 1'
2089Channel 'a' is closed
2090Channel 'a' is closed
2091```
2092
Roman Elizarov731f0ad2017-02-22 20:48:45 +03002093<!--- TEST -->
2094
Roman Elizarova84730b2017-02-22 11:58:50 +03002095There are couple of observations to make out of it.
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002096
2097First of all, `select` is _biased_ to the first clause. When several clauses are selectable at the same time,
2098the first one among them gets selected. Here, both channels are constantly producing strings, so `a` channel,
Roman Elizarova84730b2017-02-22 11:58:50 +03002099being the first clause in select, wins. However, because we are using unbuffered channel, the `a` gets suspended from
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002100time to time on its [send][SendChannel.send] invocation and gives a chance for `b` to send, too.
2101
Roman Elizarov8a5564d2017-09-06 18:48:22 +03002102The second observation, is that [onReceiveOrNull][ReceiveChannel.onReceiveOrNull] gets immediately selected when the
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002103channel is already closed.
2104
2105### Selecting to send
2106
Roman Elizarov8a5564d2017-09-06 18:48:22 +03002107Select expression has [onSend][SendChannel.onSend] clause that can be used for a great good in combination
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002108with a biased nature of selection.
2109
Roman Elizarova84730b2017-02-22 11:58:50 +03002110Let us write an example of producer of integers that sends its values to a `side` channel when
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002111the consumers on its primary channel cannot keep up with it:
2112
Roman Elizarov8b38fa22017-09-27 17:44:31 +03002113<!--- INCLUDE
2114import kotlin.coroutines.experimental.CoroutineContext
2115-->
2116
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002117```kotlin
Roman Elizarov8b38fa22017-09-27 17:44:31 +03002118fun produceNumbers(context: CoroutineContext, side: SendChannel<Int>) = produce<Int>(context) {
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002119 for (num in 1..10) { // produce 10 numbers from 1 to 10
2120 delay(100) // every 100 ms
2121 select<Unit> {
Roman Elizarova84730b2017-02-22 11:58:50 +03002122 onSend(num) {} // Send to the primary channel
2123 side.onSend(num) {} // or to the side channel
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002124 }
2125 }
2126}
2127```
2128
2129Consumer is going to be quite slow, taking 250 ms to process each number:
2130
2131```kotlin
2132fun main(args: Array<String>) = runBlocking<Unit> {
2133 val side = Channel<Int>() // allocate side channel
Roman Elizarov43e3af72017-07-21 16:01:31 +03002134 launch(coroutineContext) { // this is a very fast consumer for the side channel
Roman Elizarov86349be2017-03-17 16:47:37 +03002135 side.consumeEach { println("Side channel has $it") }
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002136 }
Roman Elizarov8b38fa22017-09-27 17:44:31 +03002137 produceNumbers(coroutineContext, side).consumeEach {
Roman Elizarov86349be2017-03-17 16:47:37 +03002138 println("Consuming $it")
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002139 delay(250) // let us digest the consumed number properly, do not hurry
2140 }
2141 println("Done consuming")
Roman Elizarov8b38fa22017-09-27 17:44:31 +03002142 coroutineContext.cancelChildren()
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002143}
2144```
2145
Roman Elizarove8d79342017-08-29 15:21:21 +03002146> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-select-03.kt)
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002147
2148So let us see what happens:
2149
Roman Elizarov731f0ad2017-02-22 20:48:45 +03002150```text
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002151Consuming 1
2152Side channel has 2
2153Side channel has 3
2154Consuming 4
2155Side channel has 5
2156Side channel has 6
2157Consuming 7
2158Side channel has 8
2159Side channel has 9
2160Consuming 10
2161Done consuming
2162```
2163
Roman Elizarov731f0ad2017-02-22 20:48:45 +03002164<!--- TEST -->
2165
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002166### Selecting deferred values
2167
Roman Elizarov8a5564d2017-09-06 18:48:22 +03002168Deferred values can be selected using [onAwait][Deferred.onAwait] clause.
Roman Elizarova84730b2017-02-22 11:58:50 +03002169Let us start with an async function that returns a deferred string value after
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002170a random delay:
2171
2172<!--- INCLUDE .*/example-select-04.kt
2173import java.util.*
2174-->
2175
2176```kotlin
Roman Elizarov66f018c2017-09-29 21:39:03 +03002177fun asyncString(time: Int) = async {
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002178 delay(time.toLong())
2179 "Waited for $time ms"
2180}
2181```
2182
Roman Elizarova84730b2017-02-22 11:58:50 +03002183Let us start a dozen of them with a random delay.
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002184
2185```kotlin
2186fun asyncStringsList(): List<Deferred<String>> {
2187 val random = Random(3)
Roman Elizarova84730b2017-02-22 11:58:50 +03002188 return List(12) { asyncString(random.nextInt(1000)) }
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002189}
2190```
2191
Roman Elizarova84730b2017-02-22 11:58:50 +03002192Now the main function awaits for the first of them to complete and counts the number of deferred values
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002193that are still active. Note, that we've used here the fact that `select` expression is a Kotlin DSL,
Roman Elizarova84730b2017-02-22 11:58:50 +03002194so we can provide clauses for it using an arbitrary code. In this case we iterate over a list
2195of deferred values to provide `onAwait` clause for each deferred value.
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002196
2197```kotlin
2198fun main(args: Array<String>) = runBlocking<Unit> {
2199 val list = asyncStringsList()
2200 val result = select<String> {
2201 list.withIndex().forEach { (index, deferred) ->
2202 deferred.onAwait { answer ->
2203 "Deferred $index produced answer '$answer'"
2204 }
2205 }
2206 }
2207 println(result)
Roman Elizarov7c864d82017-02-27 10:17:50 +03002208 val countActive = list.count { it.isActive }
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002209 println("$countActive coroutines are still active")
2210}
2211```
2212
Roman Elizarove8d79342017-08-29 15:21:21 +03002213> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-select-04.kt)
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002214
2215The output is:
2216
Roman Elizarov731f0ad2017-02-22 20:48:45 +03002217```text
Roman Elizarova84730b2017-02-22 11:58:50 +03002218Deferred 4 produced answer 'Waited for 128 ms'
Roman Elizarovd4dcbe22017-02-22 09:57:46 +0300221911 coroutines are still active
2220```
2221
Roman Elizarov731f0ad2017-02-22 20:48:45 +03002222<!--- TEST -->
2223
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002224### Switch over a channel of deferred values
2225
Roman Elizarova84730b2017-02-22 11:58:50 +03002226Let us write a channel producer function that consumes a channel of deferred string values, waits for each received
2227deferred value, but only until the next deferred value comes over or the channel is closed. This example puts together
Roman Elizarov8a5564d2017-09-06 18:48:22 +03002228[onReceiveOrNull][ReceiveChannel.onReceiveOrNull] and [onAwait][Deferred.onAwait] clauses in the same `select`:
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002229
2230```kotlin
Roman Elizarov66f018c2017-09-29 21:39:03 +03002231fun switchMapDeferreds(input: ReceiveChannel<Deferred<String>>) = produce<String> {
Roman Elizarova84730b2017-02-22 11:58:50 +03002232 var current = input.receive() // start with first received deferred value
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002233 while (isActive) { // loop while not cancelled/closed
2234 val next = select<Deferred<String>?> { // return next deferred value from this select or null
2235 input.onReceiveOrNull { update ->
2236 update // replaces next value to wait
2237 }
2238 current.onAwait { value ->
2239 send(value) // send value that current deferred has produced
2240 input.receiveOrNull() // and use the next deferred from the input channel
2241 }
2242 }
2243 if (next == null) {
2244 println("Channel was closed")
2245 break // out of loop
2246 } else {
2247 current = next
2248 }
2249 }
2250}
2251```
2252
2253To test it, we'll use a simple async function that resolves to a specified string after a specified time:
2254
2255```kotlin
Roman Elizarov66f018c2017-09-29 21:39:03 +03002256fun asyncString(str: String, time: Long) = async {
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002257 delay(time)
2258 str
2259}
2260```
2261
2262The main function just launches a coroutine to print results of `switchMapDeferreds` and sends some test
2263data to it:
2264
2265```kotlin
2266fun main(args: Array<String>) = runBlocking<Unit> {
2267 val chan = Channel<Deferred<String>>() // the channel for test
Roman Elizarov43e3af72017-07-21 16:01:31 +03002268 launch(coroutineContext) { // launch printing coroutine
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002269 for (s in switchMapDeferreds(chan))
2270 println(s) // print each received string
2271 }
2272 chan.send(asyncString("BEGIN", 100))
2273 delay(200) // enough time for "BEGIN" to be produced
2274 chan.send(asyncString("Slow", 500))
Roman Elizarova84730b2017-02-22 11:58:50 +03002275 delay(100) // not enough time to produce slow
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002276 chan.send(asyncString("Replace", 100))
Roman Elizarova84730b2017-02-22 11:58:50 +03002277 delay(500) // give it time before the last one
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002278 chan.send(asyncString("END", 500))
2279 delay(1000) // give it time to process
Roman Elizarova84730b2017-02-22 11:58:50 +03002280 chan.close() // close the channel ...
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002281 delay(500) // and wait some time to let it finish
2282}
2283```
2284
Roman Elizarove8d79342017-08-29 15:21:21 +03002285> You can get full code [here](core/kotlinx-coroutines-core/src/test/kotlin/guide/example-select-05.kt)
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002286
2287The result of this code:
2288
Roman Elizarov731f0ad2017-02-22 20:48:45 +03002289```text
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002290BEGIN
2291Replace
2292END
2293Channel was closed
2294```
2295
Roman Elizarov731f0ad2017-02-22 20:48:45 +03002296<!--- TEST -->
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002297
Roman Elizarov8db17332017-03-09 12:40:45 +03002298## Further reading
2299
2300* [Guide to UI programming with coroutines](ui/coroutines-guide-ui.md)
Roman Elizarov8a4a8e12017-03-09 19:52:58 +03002301* [Guide to reactive streams with coroutines](reactive/coroutines-guide-reactive.md)
Roman Elizarov8db17332017-03-09 12:40:45 +03002302* [Coroutines design document (KEEP)](https://github.com/Kotlin/kotlin-coroutines/blob/master/kotlin-coroutines-informal.md)
2303* [Full kotlinx.coroutines API reference](http://kotlin.github.io/kotlinx.coroutines)
2304
Roman Elizarove7e2ad12017-05-17 14:47:31 +03002305<!--- MODULE kotlinx-coroutines-core -->
Roman Elizarove0c817d2017-02-10 10:22:01 +03002306<!--- INDEX kotlinx.coroutines.experimental -->
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002307[launch]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/launch.html
2308[delay]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/delay.html
2309[runBlocking]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/run-blocking.html
Roman Elizarove82dee72017-08-18 16:49:09 +03002310[Job]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-job/index.html
Roman Elizarov8b38fa22017-09-27 17:44:31 +03002311[cancelAndJoin]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/cancel-and-join.html
Roman Elizarov88396732017-09-27 21:30:47 +03002312[Job.cancel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-job/cancel.html
2313[Job.join]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-job/join.html
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002314[CancellationException]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-cancellation-exception.html
2315[yield]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/yield.html
Roman Elizarovbff3f372017-03-01 18:12:27 +03002316[CoroutineScope.isActive]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-scope/is-active.html
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002317[CoroutineScope]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-scope/index.html
2318[run]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/run.html
2319[NonCancellable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-non-cancellable/index.html
2320[withTimeout]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/with-timeout.html
Roman Elizarov63f6ea22017-09-06 18:42:34 +03002321[withTimeoutOrNull]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/with-timeout-or-null.html
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002322[async]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/async.html
2323[Deferred]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-deferred/index.html
Roman Elizarovecda27f2017-04-06 23:06:26 +03002324[CoroutineStart.LAZY]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-start/-l-a-z-y.html
Roman Elizarovbff3f372017-03-01 18:12:27 +03002325[Deferred.await]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-deferred/await.html
2326[Job.start]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-job/start.html
Roman Elizarov419a6c82017-02-09 18:36:22 +03002327[CoroutineDispatcher]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-dispatcher/index.html
Roman Elizarov66f018c2017-09-29 21:39:03 +03002328[CommonPool]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-common-pool/index.html
Roman Elizarov43e3af72017-07-21 16:01:31 +03002329[CoroutineScope.coroutineContext]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-scope/coroutine-context.html
Roman Elizarov419a6c82017-02-09 18:36:22 +03002330[Unconfined]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-unconfined/index.html
Roman Elizarov419a6c82017-02-09 18:36:22 +03002331[newCoroutineContext]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/new-coroutine-context.html
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002332[CoroutineName]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-name/index.html
Roman Elizarov88396732017-09-27 21:30:47 +03002333[Job()]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-job.html
2334[cancelChildren]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/kotlin.coroutines.experimental.-coroutine-context/cancel-children.html
Roman Elizarove82dee72017-08-18 16:49:09 +03002335[CompletableDeferred]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-completable-deferred/index.html
Roman Elizarov8a5564d2017-09-06 18:48:22 +03002336[Deferred.onAwait]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-deferred/on-await.html
Roman Elizarovf5bc0472017-02-22 11:38:13 +03002337<!--- INDEX kotlinx.coroutines.experimental.sync -->
Roman Elizarove82dee72017-08-18 16:49:09 +03002338[Mutex]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.sync/-mutex/index.html
Roman Elizarovbff3f372017-03-01 18:12:27 +03002339[Mutex.lock]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.sync/-mutex/lock.html
2340[Mutex.unlock]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.sync/-mutex/unlock.html
Roman Elizarov88396732017-09-27 21:30:47 +03002341[withLock]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.sync/with-lock.html
Roman Elizarove0c817d2017-02-10 10:22:01 +03002342<!--- INDEX kotlinx.coroutines.experimental.channels -->
Roman Elizarove82dee72017-08-18 16:49:09 +03002343[Channel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-channel/index.html
Roman Elizarovbff3f372017-03-01 18:12:27 +03002344[SendChannel.send]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-send-channel/send.html
2345[ReceiveChannel.receive]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-receive-channel/receive.html
2346[SendChannel.close]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-send-channel/close.html
Roman Elizarova5e653f2017-02-13 13:49:55 +03002347[produce]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/produce.html
Roman Elizarov86349be2017-03-17 16:47:37 +03002348[consumeEach]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/consume-each.html
Roman Elizarov88396732017-09-27 21:30:47 +03002349[Channel()]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-channel.html
Roman Elizarovc0e19f82017-02-27 11:59:14 +03002350[actor]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/actor.html
Roman Elizarov8a5564d2017-09-06 18:48:22 +03002351[ReceiveChannel.onReceive]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-receive-channel/on-receive.html
2352[ReceiveChannel.onReceiveOrNull]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-receive-channel/on-receive-or-null.html
2353[SendChannel.onSend]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-send-channel/on-send.html
Roman Elizarovd4dcbe22017-02-22 09:57:46 +03002354<!--- INDEX kotlinx.coroutines.experimental.selects -->
2355[select]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.selects/select.html
Roman Elizarov419a6c82017-02-09 18:36:22 +03002356<!--- END -->