Kotlin Coroutines in Practice @ KotlinConf 2018

Let's see how Kotlin Coroutines are used to solve real-life concurrency and coordination problems. With coroutines we don't have to worry about shared mutable state and synchronization. We can solve the problems we face using a number of communicating coroutines, where each piece of state is confined to a single coroutine.

1. elizarov@ Roman Elizarov Kotlin Coroutines in Practice @relizarov
2. Coroutines recap
3. Coroutines are like light-weight threads
4. suspend fun main() { val jobs = List(100_000) { GlobalScope.launch { delay(5000) print(".") } } jobs.forEach { it.join() } } Coroutines are like light-weight threads
5. suspend fun main() { val jobs = List(100_000) { GlobalScope.launch { delay(5000) print(".") } } jobs.forEach { it.join() } } Coroutines are like light-weight threads
6. suspend fun main() { val jobs = List(100_000) { GlobalScope.launch { delay(5000) print(".") } } jobs.forEach { it.join() } } Coroutines are like light-weight threads
7. Quantity
8. Quantity → Quality
9. A practical challenge suspend fun downloadContent(location: Location): Content
10. fun processReferences(refs: List<Reference>) References
11. References fun processReferences(refs: List<Reference>) { for (ref in refs) { … } }
12. References Locations resolve fun processReferences(refs: List<Reference>) { for (ref in refs) { val location = ref.resolveLocation() … } }
13. References Locations Contents resolve download fun processReferences(refs: List<Reference>) { for (ref in refs) { val location = ref.resolveLocation() val content = downloadContent(location) processContent(ref, content) } }
14. References Locations Contents resolve download suspend fun processReferences(refs: List<Reference>) { for (ref in refs) { val location = ref.resolveLocation() val content = downloadContent(location) processContent(ref, content) } }
15. References Locations Contents resolve download suspend fun processReferences(refs: List<Reference>) { for (ref in refs) { val location = ref.resolveLocation() val content = downloadContent(location) processContent(ref, content) } } Sequential
16. References Locations Contents resolve download fun processReferences(refs: List<Reference>) { for (ref in refs) { val location = ref.resolveLocation() val content = downloadContent(location) processContent(ref, content) } } Parallel
17. References Locations Contents resolve download fun processReferences(refs: List<Reference>) { for (ref in refs) { val location = ref.resolveLocation() GlobalScope.launch { val content = downloadContent(location) processContent(ref, content) } } } Parallel
18. fun processReferences(refs: List<Reference>) { for (ref in refs) { val location = ref.resolveLocation() GlobalScope.launch { val content = downloadContent(location) processContent(ref, content) } } } Coroutines are cheap! What could go wrong?
19. fun processReferences(refs: List<Reference>) { for (ref in refs) { val location = ref.resolveLocation() GlobalScope.launch { val content = downloadContent(location) processContent(ref, content) } } } ref1 location1 Launch download ref2 location2 Launch download ref3 Crash! Crash!
20. ref1 location1 Launch download ref2 location2 Launch download ref3 Crash! Crash! Leak! fun processReferences(refs: List<Reference>) { for (ref in refs) { val location = ref.resolveLocation() GlobalScope.launch { val content = downloadContent(location) processContent(ref, content) } } }
21. Structured concurrency
22. fun processReferences(refs: List<Reference>)
23. suspend fun processReferences(refs: List<Reference>)
24. suspend fun processReferences(refs: List<Reference>) = coroutineScope { … }
25. suspend fun processReferences(refs: List<Reference>) = coroutineScope { for (ref in refs) { val location = ref.resolveLocation() GlobalScope.launch { val content = downloadContent(location) processContent(ref, content) } } }
26. suspend fun processReferences(refs: List<Reference>) = coroutineScope { for (ref in refs) { val location = ref.resolveLocation() launch { val content = downloadContent(location) processContent(ref, content) } } }
27. suspend fun processReferences(refs: List<Reference>) = coroutineScope { for (ref in refs) { val location = ref.resolveLocation() launch { val content = downloadContent(location) processContent(ref, content) } } } Child
28. Crash? suspend fun processReferences(refs: List<Reference>) = coroutineScope { for (ref in refs) { val location = ref.resolveLocation() launch { val content = downloadContent(location) processContent(ref, content) } } }
29. suspend fun processReferences(refs: List<Reference>) = coroutineScope { for (ref in refs) { val location = ref.resolveLocation() launch { val content = downloadContent(location) processContent(ref, content) } } } Crash?
30. suspend fun processReferences(refs: List<Reference>) = coroutineScope { for (ref in refs) { val location = ref.resolveLocation() launch { val content = downloadContent(location) processContent(ref, content) } } } Crash? cancels
31. Crash? cancels Waits for completion suspend fun processReferences(refs: List<Reference>) = coroutineScope { for (ref in refs) { val location = ref.resolveLocation() launch { val content = downloadContent(location) processContent(ref, content) } } }
32. suspend fun processReferences(refs: List<Reference>) = coroutineScope { for (ref in refs) { val location = ref.resolveLocation() launch { val content = downloadContent(location) processContent(ref, content) } } } Never leaks jobs
33. The state
34. References Contents Download process
35. Reference 1 Content Reference 2 Location Download process State
36. class Downloader { }
37. class Downloader { private val requested = mutableSetOf<Location>() }
38. class Downloader { private val requested = mutableSetOf<Location>() fun downloadReference(ref: Reference) { val location = ref.resolveLocation() … } }
39. class Downloader { private val requested = mutableSetOf<Location>() fun downloadReference(ref: Reference) { val location = ref.resolveLocation() if (requested.add(location)) { // schedule download } } }
40. class Downloader { private val requested = mutableSetOf<Location>() fun downloadReference(ref: Reference) { val location = ref.resolveLocation() if (requested.add(location)) { // schedule download } // ... wait for result ... processContent(ref, content) } }
41. class Downloader { private val requested = mutableSetOf<Location>() fun downloadReference(ref: Reference) { val location = ref.resolveLocation() if (requested.add(location)) { // schedule download } // ... wait for result ... processContent(ref, content) } } Concurrent
42. class Downloader { private val requested = mutableSetOf<Location>() fun downloadReference(ref: Reference) { val location = ref.resolveLocation() if (requested.add(location)) { // schedule download } // ... wait for result ... processContent(ref, content) } } Concurrent
43. class Downloader { private val requested = mutableSetOf<Location>() fun downloadReference(ref: Reference) { val location = ref.resolveLocation() if (requested.add(location)) { // schedule download } // ... wait for result ... processContent(ref, content) } } Shared mutable state
44. class Downloader { private val requested = mutableSetOf<Location>() fun downloadReference(ref: Reference) { val location = ref.resolveLocation() if (requested.add(location)) { // schedule download } // ... wait for result ... processContent(ref, content) } } Shared mutable state Needs Synchronization Shared + Mutable =
45. Shared Mutable State Share by Communicating
46. Synchronization Primitives Communication Primitives
47. classes coroutines
48. launch { val requested = mutableSetOf<Location>() … } Does not share mutable state
49. launch { val requested = mutableSetOf<Location>() for (ref in references) { val location = ref.resolveLocation() if (requested.add(location)) { // schedule download } // ... wait for result ... processContent(ref, content) } }
50. launch { val requested = mutableSetOf<Location>() for (ref in references) { val location = ref.resolveLocation() if (requested.add(location)) { // schedule download } // ... wait for result ... processContent(ref, content) } }
51. Channel References Downloader
52. fun CoroutineScope.downloader( references: ReceiveChannel<Reference>, ) = launch { … }
53. fun CoroutineScope.downloader( references: ReceiveChannel<Reference>, ) = launch { … }
54. fun CoroutineScope.downloader( references: ReceiveChannel<Reference>, ) = launch { … }
55. fun CoroutineScope.downloader( references: ReceiveChannel<Reference>, ) = launch { … } Convention
56. fun CoroutineScope.downloader( references: ReceiveChannel<Reference>, ) = launch { val requested = mutableSetOf<Location>() for (ref in references) { val location = ref.resolveLocation() if (requested.add(location)) { // schedule download } // ... wait for result ... processContent(ref, content) } }
57. fun CoroutineScope.downloader( references: ReceiveChannel<Reference>, ) = launch { val requested = mutableSetOf<Location>() for (ref in references) { val location = ref.resolveLocation() if (requested.add(location)) { // schedule download } // ... wait for result ... processContent(ref, content) } }
58. fun CoroutineScope.downloader( references: ReceiveChannel<Reference>, ) = launch { val requested = mutableSetOf<Location>() for (ref in references) { val location = ref.resolveLocation() if (requested.add(location)) { // schedule download } // ... wait for result ... processContent(ref, content) } }
59. fun CoroutineScope.downloader( references: ReceiveChannel<Reference>, ) = launch { val requested = mutableSetOf<Location>() for (ref in references) { val location = ref.resolveLocation() if (requested.add(location)) { // schedule download } // ... wait for result ... processContent(ref, content) } }
60. fun CoroutineScope.downloader( references: ReceiveChannel<Reference>, ) = launch { val requested = mutableSetOf<Location>() for (ref in references) { val location = ref.resolveLocation() if (requested.add(location)) { launch { … } } // ... wait for result ... processContent(ref, content) } } Coroutines are cheap! What could go wrong?
61. fun CoroutineScope.downloader( references: ReceiveChannel<Reference>, ) = launch { val requested = mutableSetOf<Location>() for (ref in references) { val location = ref.resolveLocation() if (requested.add(location)) { launch { … } } // ... wait for result ... processContent(ref, content) } } Child
62. fun CoroutineScope.downloader( references: ReceiveChannel<Reference>, ) = launch { val requested = mutableSetOf<Location>() for (ref in references) { val location = ref.resolveLocation() if (requested.add(location)) { launch { … } } // ... wait for result ... processContent(ref, content) } } Coroutines are cheap! But the work they do…
63. Limiting concurrency
64. Worker 1 Worker 2 Worker 3 Worker N Worker pool … References Downloader references locations
65. fun CoroutineScope.downloader( references: ReceiveChannel<Reference>, locations: SendChannel<Location> )
66. fun CoroutineScope.downloader( references: ReceiveChannel<Reference>, locations: SendChannel<Location> ) = launch { val requested = mutableSetOf<Location>() for (ref in references) { val location = ref.resolveLocation() if (requested.add(location)) { locations.send(location) } } }
67. fun CoroutineScope.worker( locations: ReceiveChannel<Location> )
68. fun CoroutineScope.worker( locations: ReceiveChannel<Location> )
69. fun CoroutineScope.worker( locations: ReceiveChannel<Location> ) = launch { for (loc in locations) { val content = downloadContent(loc) processContent(ref, content) } }
70. fun CoroutineScope.worker( locations: ReceiveChannel<Location> ) = launch { for (loc in locations) { val content = downloadContent(loc) processContent(ref, content) } } Fan-out
71. fun CoroutineScope.worker( locations: ReceiveChannel<Location> ) = launch { for (loc in locations) { val content = downloadContent(location) processContent(ref, content) } }
72. fun CoroutineScope.worker( locations: ReceiveChannel<Location> ) = launch { for (loc in locations) { val content = downloadContent(loc) processContent(ref, content) } }
73. fun CoroutineScope.worker( locations: ReceiveChannel<Location> ) = launch { for (loc in locations) { val content = downloadContent(loc) processContent(ref, content) } }
74. Worker 1 Worker 2 Worker 3 Worker N … References Downloader
75. Worker 1 Worker 2 Worker 3 Worker N … References Downloader Refs ↔ Locs location & content
76. data class LocContent(val loc: Location, val content: Content)
77. data class LocContent(val loc: Location, val content: Content) fun CoroutineScope.worker( locations: ReceiveChannel<Location>, contents: SendChannel<LocContent> )
78. data class LocContent(val loc: Location, val content: Content) fun CoroutineScope.worker( locations: ReceiveChannel<Location>, contents: SendChannel<LocContent> ) = launch { for (loc in locations) { val content = downloadContent(loc) contents.send(LocContent(loc, content)) } }
79. Worker 1 Worker 2 Worker 3 Worker N … References Downloader locations contents
80. fun CoroutineScope.downloader( references: ReceiveChannel<Reference>, locations: SendChannel<Location>, contents: ReceiveChannel<LocContent> )
81. fun CoroutineScope.downloader( references: ReceiveChannel<Reference>, locations: SendChannel<Location>, contents: ReceiveChannel<LocContent> ) = launch { val requested = mutableSetOf<Location>() for (ref in references) { val location = ref.resolveLocation() if (requested.add(location)) { locations.send(location) } } } Hmm….
82. Select
83. select { references.onReceive { ref -> … } contents.onReceive { (loc, content) -> … } }
84. select { references.onReceive { ref -> … } contents.onReceive { (loc, content) -> … } }
85. select<Unit> { references.onReceive { ref -> … } contents.onReceive { (loc, content) -> … } }
86. launch { val requested = mutableMapOf<Location, MutableList<Reference>>() … }
87. launch { val requested = mutableMapOf<Location, MutableList<Reference>>() while (true) { select<Unit> { references.onReceive { ref -> … } contents.onReceive { (loc, content) -> … } } } }
88. launch { val requested = mutableMapOf<Location, MutableList<Reference>>() while (true) { select<Unit> { references.onReceive { ref -> … } contents.onReceive { (loc, content) -> … } } } }
89. launch { val requested = mutableMapOf<Location, MutableList<Reference>>() while (true) { select<Unit> { references.onReceive { ref -> val loc = ref.resolveLocation() … } contents.onReceive { (loc, content) -> … } } } }
90. launch { val requested = mutableMapOf<Location, MutableList<Reference>>() while (true) { select<Unit> { references.onReceive { ref -> val loc = ref.resolveLocation() val refs = requested[loc] … } contents.onReceive { (loc, content) -> … } } } }
91. launch { val requested = mutableMapOf<Location, MutableList<Reference>>() while (true) { select<Unit> { references.onReceive { ref -> val loc = ref.resolveLocation() val refs = requested[loc] if (refs == null) { requested[loc] = mutableListOf(ref) locations.send(loc) } } contents.onReceive { (loc, content) -> … } } } }
92. launch { val requested = mutableMapOf<Location, MutableList<Reference>>() while (true) { select<Unit> { references.onReceive { ref -> val loc = ref.resolveLocation() val refs = requested[loc] if (refs == null) { requested[loc] = mutableListOf(ref) locations.send(loc) } else { refs.add(ref) } } contents.onReceive { (loc, content) -> … } } } }
93. launch { val requested = mutableMapOf<Location, MutableList<Reference>>() while (true) { select<Unit> { references.onReceive { ref -> val loc = ref.resolveLocation() val refs = requested[loc] if (refs == null) { requested[loc] = mutableListOf(ref) locations.send(loc) } else { refs.add(ref) } } contents.onReceive { (loc, content) -> … } } } } No concurrency No synchronization
94. launch { val requested = mutableMapOf<Location, MutableList<Reference>>() while (true) { select<Unit> { references.onReceive { ref -> … } contents.onReceive { (loc, content) -> val refs = requested.remove(loc)!! for (ref in refs) { processContent(ref, content) } } } } }
95. Putting it all together
96. Worker 1 Worker 2 Worker 3 Worker N … References Downloader locations contents references
97. Worker 1 Worker 2 Worker 3 Worker N … References Downloader locations contents references
98. fun CoroutineScope.processReferences( references: ReceiveChannel<Reference> )
99. fun CoroutineScope.processReferences( references: ReceiveChannel<Reference> )
100. fun CoroutineScope.processReferences( references: ReceiveChannel<Reference> ) { val locations = Channel<Location>() val contents = Channel<LocContent>() repeat(N_WORKERS) { worker(locations, contents) } downloader(references, locations, contents) }
101. fun CoroutineScope.processReferences( references: ReceiveChannel<Reference> ) { val locations = Channel<Location>() val contents = Channel<LocContent>() repeat(N_WORKERS) { worker(locations, contents) } downloader(references, locations, contents) }
102. Worker 1 Worker 2 Worker 3 Worker N … References Downloader locations contents references
103. Worker 1 Worker 2 Worker 3 Worker N … References Downloader locations contents references processReferences
104. Worker 1 Worker 2 Worker 3 Worker N … References Downloader locations contents references processReferences Patterns everywhere Worker pool Actor
105. fun CoroutineScope.processReferences(…)
106. Root CoroutineScope
107. class SomethingWithLifecycle { }
108. class SomethingWithLifecycle : CoroutineScope { }
109. class SomethingWithLifecycle : CoroutineScope { override val coroutineContext: CoroutineContext get() = … }
110. class SomethingWithLifecycle : CoroutineScope { private val job = Job() override val coroutineContext: CoroutineContext get() = … }
111. class SomethingWithLifecycle : CoroutineScope { private val job = Job() fun dispose() { … } override val coroutineContext: CoroutineContext get() = … }
112. class SomethingWithLifecycle : CoroutineScope { private val job = Job() fun close() { … } override val coroutineContext: CoroutineContext get() = … }
113. class SomethingWithLifecycle : CoroutineScope { private val job = Job() fun close() { job.cancel() } override val coroutineContext: CoroutineContext get() = … }
114. class SomethingWithLifecycle : CoroutineScope { private val job = Job() fun close() { job.cancel() } override val coroutineContext: CoroutineContext get() = job }
115. class SomethingWithLifecycle : CoroutineScope { private val job = Job() fun close() { job.cancel() } override val coroutineContext: CoroutineContext get() = job + Dispatchers.Main }
116. class SomethingWithLifecycle : CoroutineScope { … override val coroutineContext: CoroutineContext get() = job + Dispatchers.Main fun doSomething() { } }
117. class SomethingWithLifecycle : CoroutineScope { … override val coroutineContext: CoroutineContext get() = job + Dispatchers.Main fun doSomething() { launch { … } } }
118. class SomethingWithLifecycle : CoroutineScope { … override val coroutineContext: CoroutineContext get() = job + Dispatchers.Main fun doSomething() { processReferences(references) } } Never leak any coroutines
119. suspend vs scope
120. suspend fun downloadContent(location: Location): Content
121. suspend fun downloadContent(location: Location): Content Does something long & waits for it to complete without blocking
122. suspend fun downloadContent(location: Location): Content fun CoroutineScope.processReferences(…)
123. suspend fun downloadContent(location: Location): Content fun CoroutineScope.processReferences(…) Launches new coroutines & quickly returns, does not wait for them
124. Takeaway
125. Coroutines are like light-weight threads
126. Coroutines are NOT like threads
127. Coroutines are NOT like threads Rethink the way you structure your code
128. Thank you Any questions? elizarov@ Roman Elizarov @relizarov