Kotlin Coroutines, Threads and Communication

As the async and possibly parallel programming becomes easier, the risk of sharing mutable variables between coroutines arises.

When the boundaries are abstracted away, we should rely on safe ways to communicate between our coroutines. In this session, I will show how coroutines work with threads and both safe and unsafe forms of communication using coroutines.

Transcript

1. Coroutines, Threads and Communication Bob Dahlberg Mobile Lead Developer
2. Coroutines are lightweight threads “
3. Lightweight Coroutines Let’s test lightweight first.
4. Lightweight Coroutines Let’s test lightweight first. fun main() = runBlocking<Unit> { repeat(100_000) { launch { !// creates a coroutine println(“On thread !-> $thread”) } } }
5. Lightweight Coroutines Let’s test lightweight first. repeat(100_000) { launch { !// creates a coroutine println(“On thread !-> $thread”) } } !// on main thread
6. Lightweight Coroutines Let’s test lightweight first. !// on main thread repeat(100_000) { thread { !// creates a thread println(“On thread !-> $thread”) } }
7. Lightweight Dispatchers Coroutines Let’s test lightweight first. !// on main thread repeat(100_000) { launch(Dispatchers.Default) { println(“On thread !-> $thread”) } }
8. Lightweight Dispatchers Coroutines Let’s test lightweight first. !// on main thread repeat(100_000) { launch(Dispatchers.IO) { println(“On thread !-> $thread”) } }
9. Lightweight Dispatchers Thread safety Coroutines Let’s take a detour around thread safety
10. Lightweight Dispatchers Thread safety Coroutines Let’s take a detour around thread safety !// on main thread var i = 0 repeat(100_000) { launch(Dispatchers.Default) { i!++ } } println(“Result !-> $i”)
11. Lightweight Dispatchers Thread safety Coroutines Let’s take a detour around thread safety !// on main thread var i = 0 repeat(100_000) { launch(Dispatchers.Default) { i!++ } } println(“Result !-> $i”) Output: Result !-> 85325
12. Lightweight Dispatchers Thread safety Coroutines Let’s take a detour around thread safety !// on main thread var i = 0 repeat(100_000) { launch(Dispatchers.IO) { i!++ } } println(“Result !-> $i”) Output: Result !-> 78332
13. Lightweight Dispatchers Thread safety Coroutines Let’s take a detour around thread safety !// on main thread var i = 0 repeat(100_000) { launch { i!++ } } println(“Result !-> $i”) Output: Result !-> 100000
14. Lightweight Dispatchers Thread safety Coroutines Let’s take a detour around thread safety !// on main thread var i = 0 launch(Dispatchers.Default) { repeat(100_000) { launch { i!++ } } } println(“Result !-> $i”) Output: Result !-> 92342
15. Lightweight Dispatchers Thread safety Threads? Coroutines Coroutines are lightweight threads lightweight
16. Lightweight Dispatchers Thread safety Threads? Coroutines Coroutines are threads
17. Lightweight Dispatchers Thread safety Threads? Coroutines Coroutines are threads !// on main thread launch(Dispatchers.Unconfined) { println(“A1. On thread $thread”) delay(200) println(“A2. On thread $thread”) } Output: A1. On thread main A2. On thread worker-1
18. Lightweight Dispatchers Thread safety Threads? Coroutines !// on main thread launch(Dispatchers.IO) { println(“A1. On thread $thread”) switchContext() println(“A2. On thread $thread”) } suspend fun switchContext() { withContext(Dispatchers.Default) { println(“B1. Switching $thread”) } } Output: A1. On thread worker-1 B1. Switching worker-1 A2. On thread worker-1
19. Lightweight Dispatchers Thread safety Threads? Coroutines !// on main thread launch(Dispatchers.IO) { println(“A1. On thread $thread”) switchContext() println(“A2. On thread $thread”) } suspend fun switchContext() { withContext(Dispatchers.Default) { println(“B1. Switching $thread”) } } Output: A1. On thread worker-1 B1. Switching worker-3 A2. On thread worker-3
20. Lightweight Dispatchers Thread safety Threads? Coroutines !// on main thread val local = ThreadLocal<String?>() launch(Dispatchers.IO) { local.set(“IO") println(“A1. ${local.get()}”) switchContext() println(“A2. ${local.get()}”) } suspend fun switchContext() { withContext(Dispatchers.Default) { println(“B1. ${local.get()}”) local.set(“Default") } } Output: A1. IO B1. IO A2. Default
21. Lightweight Dispatchers Thread safety Threads? Coroutines !// on main thread val local = ThreadLocal<String?>() launch(Dispatchers.IO) { local.set(“IO") println(“A1. ${local.get()}”) switchContext() println(“A2. ${local.get()}”) } suspend fun switchContext() { withContext(Dispatchers.Default) { println(“B1. ${local.get()}”) local.set(“Default") } } Output: A1. IO B1. null A2. Default
22. Lightweight Dispatchers Thread safety Threads? Coroutines !// on main thread @Synchronized fun critical() { println(“Start $thread”) Thread.sleep(100) println(“Stop $thread”) } repeat(2) { thread { critical() } } Output: Start worker-1 Stop worker-1 Start worker-3 Stop worker-3
23. Lightweight Dispatchers Thread safety Threads? Coroutines !// on main thread @Synchronized suspend fun critical() { println(“Start $thread”) delay(100) println(“Stop $thread”) } repeat(2) { launch(Dispatchers.Default) { critical() } } Output: Start worker-1 Start worker-3 Stop worker-1 Stop worker-3
24. Lightweight Dispatchers Thread safety Threads? Coroutines !// on main thread suspend fun critical() { synchronized(obj) { println(“Start $thread”) delay(100) println(“Stop $thread”) } } repeat(2) { launch(Dispatchers.Default) { critical() } } Output: Start worker-1 Start worker-3 Stop worker-1 Stop worker-3 The 'delay' suspension point is inside a critical section
25. Key take aways
26. coroutines !== lightweight
27. coroutines !!= threads
28. dispatchers !!= thread pools
29. How to communicate coroutine-style
30. Deferred Communicate Deferred is a non-blocking cancelable future
 val result: Deferred<Any> = async { fetchChannels() } println(“Deferred !-> ${result.await()}”)
31. Deferred Communicate Deferred is a non-blocking cancelable future
 val result = CompletableDeferred<Json>() launch(Dispatchers.IO) { result.complete(slowFetch()) } println(“Deferred !-> ${result.await()}”)
32. Deferred Mutex Communicate Mutex is Kotlins Mutual Exclusion
 val mutex = Mutex() suspend fun critical() { mutex.withLock { println(“Start $thread”) delay(100) println(“Stop $thread”) } } repeat(2) { launch(Dispatchers.Default) { critical() } }
33. Deferred Mutex Channel Communicate Channels are synchronization primitives for streams
 suspend fun compete(name:String): String { delay(nextLong(5000)) return name } val race = Channel<String>() launch(…) { race.send(compete("Bob")) } launch(…) { race.send(compete(“Emilie")) } launch(…) { race.send(compete(“Charlie")) } launch(…) { race.send(compete(“Filippa")) } launch(Dispatchers.Default) { repeat(4) { println(“Name: ${race.receive()}") } race.close() }
34. Deferred Mutex Channel Communicate Channels are synchronization primitives for streams
 val ch: Channel<Int> = produce { repeat(30) { send(it) } } repeat(3) { id !-> launch(Dispatchers.Default) { for(msg in ch) { println(“$id !-> $msg”) } } }
35. Deferred Mutex Channel Communicate Channels are synchronization primitives for streams
 Channel<String>(7) !// BUFFERED Channel<String>(Channel.UNLIMITED) Channel<String>(Channel.CONFLATED) Channel<String>(Channel.RENDEZVOUS)
36. Deferred Mutex Channel Flow Communicate Kotlins take on reactive streams
 val example = flow { for(i in 1!..10) { println(“Flow on !-> ${thread()}”) emit(i) } } !// on main thread example.filter { it !>= 5 } .map { it * 2 } .collect { println(“Collect on !-> ${thread()}”) }
37. Deferred Mutex Channel Flow Communicate Kotlins take on reactive streams
 val example = flow { for(i in 1!..10) { println(“Flow on !-> ${thread()}”) emit(i) } }.flowOn(Dispatchers.Default) !// on main thread example.filter { it !>= 5 } .map { it * 2 } .collect { println(“Collect on !-> ${thread()}”) }
38. Deferred Mutex Channel Flow Communicate Kotlins take on reactive streams
 val example = flow { for(i in 1!..10) { println(“Flow on !-> ${thread()}”) emit(i) } } !// on main thread example.filter { it !>= 5 } .map { it * 2 } .flowOn(Dispatchers.Default) .collect { println(“Collect on !-> ${thread()}”) }
39. Deferred Mutex Channel Flow Communicate Kotlins take on reactive streams
 val example = flow { for(i in 1!..10) { println(“Flow on !-> ${thread()}”) emit(i) } } !// on main thread example.filter { it !>= 5 } .flowOn(Dispatchers.IO) .map { it * 2 } .flowOn(Dispatchers.Default) .collect { println(“Collect on !-> ${thread()}”) }
40. Thank you!
41. Questions? Bob Dahlberg bob@qvik.com medium.com/dahlbergbob @mr_bob This Presentation tiny.cc/fosdem-bob speakerdeck.com/bobdahlberg