Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Kotlin Coroutines in Practice @ KotlinConf 2018

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.

Roman Elizarov

October 05, 2018
Tweet

More Decks by Roman Elizarov

Other Decks in Programming

Transcript

  1. suspend fun main() { val jobs = List(100_000) { GlobalScope.launch

    { delay(5000) print(".") } } jobs.forEach { it.join() } } Coroutines are like light-weight threads
  2. suspend fun main() { val jobs = List(100_000) { GlobalScope.launch

    { delay(5000) print(".") } } jobs.forEach { it.join() } } Coroutines are like light-weight threads
  3. suspend fun main() { val jobs = List(100_000) { GlobalScope.launch

    { delay(5000) print(".") } } jobs.forEach { it.join() } } Coroutines are like light-weight threads
  4. 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) } }
  5. 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) } }
  6. 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
  7. 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
  8. 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
  9. 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?
  10. 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!
  11. 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) } } }
  12. suspend fun processReferences(refs: List<Reference>) = coroutineScope { for (ref in

    refs) { val location = ref.resolveLocation() GlobalScope.launch { val content = downloadContent(location) processContent(ref, content) } } }
  13. suspend fun processReferences(refs: List<Reference>) = coroutineScope { for (ref in

    refs) { val location = ref.resolveLocation() launch { val content = downloadContent(location) processContent(ref, content) } } }
  14. suspend fun processReferences(refs: List<Reference>) = coroutineScope { for (ref in

    refs) { val location = ref.resolveLocation() launch { val content = downloadContent(location) processContent(ref, content) } } } Child
  15. Crash? suspend fun processReferences(refs: List<Reference>) = coroutineScope { for (ref

    in refs) { val location = ref.resolveLocation() launch { val content = downloadContent(location) processContent(ref, content) } } }
  16. suspend fun processReferences(refs: List<Reference>) = coroutineScope { for (ref in

    refs) { val location = ref.resolveLocation() launch { val content = downloadContent(location) processContent(ref, content) } } } Crash?
  17. 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
  18. 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) } } }
  19. 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
  20. class Downloader { private val requested = mutableSetOf<Location>() fun downloadReference(ref:

    Reference) { val location = ref.resolveLocation() if (requested.add(location)) { // schedule download } } }
  21. 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) } }
  22. 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
  23. 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
  24. 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
  25. 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 =
  26. 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) } }
  27. 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) } }
  28. 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) } }
  29. 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) } }
  30. 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) } }
  31. 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) } }
  32. 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?
  33. 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
  34. 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…
  35. Worker 1 Worker 2 Worker 3 Worker N Worker pool

    … References Downloader references locations
  36. 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) } } }
  37. fun CoroutineScope.worker( locations: ReceiveChannel<Location> ) = launch { for (loc

    in locations) { val content = downloadContent(loc) processContent(ref, content) } }
  38. fun CoroutineScope.worker( locations: ReceiveChannel<Location> ) = launch { for (loc

    in locations) { val content = downloadContent(loc) processContent(ref, content) } } Fan-out
  39. fun CoroutineScope.worker( locations: ReceiveChannel<Location> ) = launch { for (loc

    in locations) { val content = downloadContent(location) processContent(ref, content) } }
  40. fun CoroutineScope.worker( locations: ReceiveChannel<Location> ) = launch { for (loc

    in locations) { val content = downloadContent(loc) processContent(ref, content) } }
  41. fun CoroutineScope.worker( locations: ReceiveChannel<Location> ) = launch { for (loc

    in locations) { val content = downloadContent(loc) processContent(ref, content) } }
  42. Worker 1 Worker 2 Worker 3 Worker N … References

    Downloader Refs ↔ Locs location & content
  43. data class LocContent(val loc: Location, val content: Content) fun CoroutineScope.worker(

    locations: ReceiveChannel<Location>, contents: SendChannel<LocContent> )
  44. 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)) } }
  45. 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….
  46. launch { val requested = mutableMapOf<Location, MutableList<Reference>>() while (true) {

    select<Unit> { references.onReceive { ref -> … } contents.onReceive { (loc, content) -> … } } } }
  47. launch { val requested = mutableMapOf<Location, MutableList<Reference>>() while (true) {

    select<Unit> { references.onReceive { ref -> … } contents.onReceive { (loc, content) -> … } } } }
  48. launch { val requested = mutableMapOf<Location, MutableList<Reference>>() while (true) {

    select<Unit> { references.onReceive { ref -> val loc = ref.resolveLocation() … } contents.onReceive { (loc, content) -> … } } } }
  49. 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) -> … } } } }
  50. 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) -> … } } } }
  51. 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) -> … } } } }
  52. 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
  53. 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) } } } } }
  54. Worker 1 Worker 2 Worker 3 Worker N … References

    Downloader locations contents references
  55. Worker 1 Worker 2 Worker 3 Worker N … References

    Downloader locations contents references
  56. fun CoroutineScope.processReferences( references: ReceiveChannel<Reference> ) { val locations = Channel<Location>()

    val contents = Channel<LocContent>() repeat(N_WORKERS) { worker(locations, contents) } downloader(references, locations, contents) }
  57. fun CoroutineScope.processReferences( references: ReceiveChannel<Reference> ) { val locations = Channel<Location>()

    val contents = Channel<LocContent>() repeat(N_WORKERS) { worker(locations, contents) } downloader(references, locations, contents) }
  58. Worker 1 Worker 2 Worker 3 Worker N … References

    Downloader locations contents references
  59. Worker 1 Worker 2 Worker 3 Worker N … References

    Downloader locations contents references processReferences
  60. Worker 1 Worker 2 Worker 3 Worker N … References

    Downloader locations contents references processReferences Patterns everywhere Worker pool Actor
  61. class SomethingWithLifecycle : CoroutineScope { private val job = Job()

    override val coroutineContext: CoroutineContext get() = … }
  62. class SomethingWithLifecycle : CoroutineScope { private val job = Job()

    fun dispose() { … } override val coroutineContext: CoroutineContext get() = … }
  63. class SomethingWithLifecycle : CoroutineScope { private val job = Job()

    fun close() { … } override val coroutineContext: CoroutineContext get() = … }
  64. class SomethingWithLifecycle : CoroutineScope { private val job = Job()

    fun close() { job.cancel() } override val coroutineContext: CoroutineContext get() = … }
  65. class SomethingWithLifecycle : CoroutineScope { private val job = Job()

    fun close() { job.cancel() } override val coroutineContext: CoroutineContext get() = job }
  66. class SomethingWithLifecycle : CoroutineScope { private val job = Job()

    fun close() { job.cancel() } override val coroutineContext: CoroutineContext get() = job + Dispatchers.Main }
  67. class SomethingWithLifecycle : CoroutineScope { … override val coroutineContext: CoroutineContext

    get() = job + Dispatchers.Main fun doSomething() { launch { … } } }
  68. class SomethingWithLifecycle : CoroutineScope { … override val coroutineContext: CoroutineContext

    get() = job + Dispatchers.Main fun doSomething() { processReferences(references) } } Never leak any coroutines