Introduction to kotlin coroutines

Introduction to Kotlin coroutines
- Kotlin 개발자를 대상으로 Kotlin coroutines에 대하여 설명합니다.

1. Svetlana Isakova
Kotlin Coroutines

2. Kotlin Coroutines
old new way to do
asynchronous programming

3. Inspired by
•async / await in C#
•coroutines and channels in Go
•and many more

4. Agenda
•Motivation: avoid callbacks
•suspend Functions
•Coroutines
•Structural Concurrency
•Channels

5. async/await

6. Motivation
time consuming
operation
val image = loadImage(url)
setImage(image)

7. Solution 1: callbacks
loadImageAsync().whenComplete { image ->
runOnUiThread {
setImage(image)
}
}

8. Solution 2: async/await
async(Main) {
val image = loadImageAsync(url).await()
setImage(image)
}

9. val image = loadImageAsync(url).await()
setImage(image)
val image = loadImage(url)
setImage(image)
No callbacks!

10. async Task ProcessImage(String url)
{
var image = await LoadImage(url);
SetImage(image);
}
fun processImage() = async {
val image = loadImageAsync().await()
setImage(image)
}
async/await

11. coroutines
async/await
Language
Library
async and await are functions
defined in the standard library

12. Programming with suspend
functions

13. Example: simple consecutive logic
Use authentication service:
fun login(email: String): UserId
Send a request to a remote data base:
fun load(id: UserId): User
Show the results:
fun show(user: User)

14. Example: simple consecutive logic
fun login(email: String): UserId
fun load(id: UserId): User
fun show(user: User)
fun showUser(email: String) {
val id = login(email)
val user = load(id)
show(user)
}
Simple, but wrong…

15. Rewrite with CompletableFuture
fun login(email: String): CompletableFuture<UserId>
fun load(id: UserId): CompletableFuture<User>
fun show(user: User)
fun showUser(email: String) {
login(email)
.thenCompose { load(it) }
.thenAccept { show(it) }
}
Looks complicated…

16. Rewrite with RxJava
fun login(email: String): Single<UserId>
fun load(id: UserId): Single<User>
fun show(user: User)
fun showUser(email: String) {
login(email)
.flatMap { load(it) }
.doOnSuccess { show(it) }
.subscribe()
}
Looks even more complicated…

17. Using async/await in Kotlin
fun login(email: String): Deferred<UserId>
fun load(id: UserId): Deferred<User>
fun show(user: User)
fun showUser(email: String) = async {
val id = login(email).await()
val user = load(id).await()
show(user)
}
runs the code asynchronously
awaits the result of the
asynchronous computation

18. Using async/await in Kotlin
fun login(email: String): Deferred<UserId>
fun load(id: UserId): Deferred<User>
fun show(user: User)
fun showUser(email: String) = async {
val id = login(email).await()
val user = load(id).await()
show(user)
}
Looks better…

19. Rewrite with suspend functions
suspend fun login(email: String): UserId
suspend fun load(id: UserId): User
fun show(user: User)
suspend fun showUser(email: String) {
val id = login(email)
val user = load(id)
show(user)
}
Looks exactly like the initial code!

20. suspend function
function that can be suspended

21. fun test(email: String) {
showUser(email)
}
Error: Suspend function 'showUser'
should be called only from a coroutine
or another suspend function
Q: Where can I call suspend functions?

22. Q: Where can I call suspend functions?
A: Inside coroutines and other suspend functions.

23. Coroutines

24. Thread Coroutine
Blocking
thread
Suspending
coroutine
From threads to coroutines

25. Coroutine
computation that can be suspended
thread

coroutine

26. Coroutine
computation that can be suspended
thread

suspended:

coroutine

27. Coroutine
computation that can be suspended
thread

suspended:

coroutine

28. Coroutine
computation that can be suspended
thread

suspended:

coroutine

29. Coroutine
computation that can be suspended
thread

suspended:


coroutine

Thread is not blocked!

30. Q: How to create a coroutine?
A: Use of so-called “coroutine builders”.

31. -Library functions
-To start a new computation asynchronously:
async { … }
-To start a new computation in a blocking way (often an entry-point):
runBlocking { … }
Coroutine builders

32. fun processImage() = async {
val image = loadImageAsync().await()
setImage(image)
}
fun loadImageAsync() = async {
/* do the work */
}
Simple “load image” example

33. async creates a new coroutine:
starts a new asynchronous computation
fun loadImageAsync() = async {
/* do the work */
}
loadImageAsync

34. suspension point
fun processImage() = async {
val image = loadImageAsync().await()
setImage(image)
}
await can suspend a coroutine

35. fun loadImageAsync(): Deferred<Image> = async {
/* do the work */
}
interface Deferred<out T> {

suspend fun await(): T

}
await is defined as a suspend function

36. await suspends coroutine
processImage
loadImageAsync
fun processImage() = async {
val deferred = loadImageAsync()
val image = deferred.await()
setImage(image)
}

37. await suspends coroutine
processImage
loadImageAsync
await
fun processImage() = async {
val deferred = loadImageAsync()
val image = deferred.await()
setImage(image)
}

38. await suspends coroutine
processImage
loadImageAsync
await
fun processImage() = async {
val deferred = loadImageAsync()
val image = deferred.await()
setImage(image)
}
suspended:

39. await suspends coroutine
processImage
loadImageAsync
await
fun processImage() = async {
val deferred = loadImageAsync()
val image = deferred.await()
setImage(image)
}
suspended:

40. Coroutine can have many suspension points
suspend fun showUser(email: String) {
val id = login(email)
val user = load(id)
show(user)
}
suspended: login() load()

41. Suspension points
suspend fun showUser(email: String) {
val id = login(email)
val user = load(id)
show(user)
}
calls of other suspend functions

42. Q: Which coroutine gets suspended when a
suspend function is called?
A: The one that contains this suspend function.

43. Outer coroutine

suspended:


async {
...
showUser()
...
}

44. Call stack of a coroutine
async
showUser
loadUser
await / library call
suspendCoroutine
suspendCoroutine call
is the language
mechanism to suspend
a given coroutine

45. Call stack of a coroutine
async
showUser
loadUser
await / library call
suspendCoroutine
application layer
library layer
language support

46. Suspended coroutine
async
showUser
loadUser
await / library call
suspendCoroutine
- suspended coroutine is
stored on the heap
- the call stack and values
of all the local variables
are saved
- only one object is used
to store a coroutine

47. Resumed coroutine
async
showUser
loadUser
await / library call
suspendCoroutine
-the call stack is restored
-the execution of the
coroutine continues

48. suspend fun foo(): Int
suspend fun foo(continuation: Continuation<Int> ): Int
“Callbacks” under the hood
Continuation is a generic callback interface:
interface Continuation<in T> {
val context: CoroutineContext
fun resume(value: T)
fun resumeWithException(exception: Throwable)
}
Each suspend function has a hidden parameter:

49. Q: On which thread does the coroutine
resume?
A: You specify that.

50. Run new or resumed coroutine on a thread from the thread pool:
async(Dispatchers.Default) { ... }
Run new or resumed coroutine on the main thread:
async(Dispatchers.Main) { ... }
…
Specify the context

51. Run Schedule new or resumed coroutine on a thread from the thread
pool:
async(Dispatchers.Default) { ... }
Run Schedule new or resumed coroutine on the main thread:
async(Dispatchers.Main) { ... }
…
Specify the context

52. async {
async { … }
}
Coroutines can be nested

53. Q: Is there any relationship between
parent and child coroutines?
A: Yes.

54. Structural concurrency

55. fun overlay(image1: Image, image2: Image): Image
suspend fun loadAndOverlay() {

val first = async { loadImage("green") }

val second = async { loadImage("red") }

return overlay(first.await(), second.await()) 

}
Two asynchronous coroutines
loadAndOverlay
loadImage
loadImage

56. Q: What happens if an exception is thrown
inside the first child coroutine (during an image
loading)?
A: The second coroutine leaks!

57. Problem: leaking coroutine
✗
fails
leaks!!!
✗
fails
fun overlay(image1: Image, image2: Image): Image
suspend fun loadAndOverlay() {

val first = async { loadImage("green") }

val second = async { loadImage("red") }

return overlay(first.await(), second.await()) 

}

58. Solution: introducing local scope
suspend fun loadAndOverlay(): Image =
coroutineScope {
val first = async { loadImage("green") }
val second = async { loadImage("red") }
overlay(first.await(), second.await())
}
✗
fails
catches exception
✗
cancelled

59. suspend fun loadAndOverlay(): Image =
coroutineScope {
val first = async { loadImage("green") }
val second = async { loadImage("red") }
overlay(first.await(), second.await())
}
✗
fails
✗
fails
✗
cancelled
Solution: introducing local scope

60. Coroutine scope
•waits for completion of all the child
coroutines inside this scope
•cancels all the child coroutines if it
gets cancelled (explicitly or by
catching an exception from a child
coroutine)

61. •You can only start a new coroutine inside a scope:
Enforcing structure
coroutineScope {
async { ... }
}
GlobalScope.async { ... }

62. •Each coroutine has the corresponding scope
GlobalScope.async {
// started in the scope of outer coroutine:
this.async { ... }
}
Enforcing structure

63. Structured concurrency
•The lifespan of a coroutine is constrained
by a lifespan of the parent scope

64. Q: How to share information between different
coroutines?
A: Share by communicating (like in Go).

65. Channels

66. Channels
used for synchronization
communication
between coroutines

67. “Share by communicating”
Shared

Mutable State
Share by
Communicating
Synchronization

Primitives
Communication
Primitives

68. channel coroutine #2coroutine #1
send receive
Channel

69. channel
consumer #1
producer #1 send
receive
producer #2
producer #N
consumer #M
...
...
Producer-consumer problem

70. Send & Receive “views” for the same channel
interface SendChannel<in E> {
suspend fun send(element: E)
fun close()
}
interface ReceiveChannel<out E> {
suspend fun receive(): E
}
interface Channel<E> : SendChannel<E>, ReceiveChannel<E>

71. send receive
...unbuffered
buffered
send receive
“rendezvous”
send receive
Types of Channels
conflated

72. “Rendezvous” channel semantics
•An element is transferred from sender to receiver only when
send and receive invocations meet in time (“rendezvous”)
•send suspends until another coroutine invokes receive
•receive suspends until another coroutine invokes send

73. consumer #1
producer #1
send tasks
receive tasks
consumer #2
Producer-consumer problem

74. Producer-consumer solution: producer
val channel = Channel<Task>()
async {
channel.send(Task( "task1"))
channel.send(Task( "task2"))
channel.close()
}
producer

75. Producer-consumer solution: consumers
val channel = Channel<Task>()
...
async { worker(channel) }
async { worker(channel) }
consumer #1
suspend fun worker(channel: Channel<Task>) {
val task = channel.receive()
processTask(task)
}
consumer #2

76. Producer-consumer solution
val channel = Channel<Task>()
val task =
channel.receive()
processTask(task)
receive

77. receive
Producer-consumer solution
val channel = Channel<Task>()
waiting for “send”
val task =
channel.receive()
processTask(task)

78. send(task1)
Producer-consumer solution
val channel = Channel<Task>()
receive
waiting for “send”
val task =
channel.receive()
processTask(task)
channel.send(task1)
channel.send(task2)
channel.close()

79. send(task1)
Producer-consumer solution
val channel = Channel<Task>()
receive
Rendezvous! waiting for “send”
val task =
channel.receive()
processTask(task)
channel.send(task1)
channel.send(task2)
channel.close()

80. Producer-consumer solution
val channel = Channel<Task>()
send(task1) receive
Rendezvous!
val task =
channel.receive()
processTask(task)
processing task1
channel.send(task1)
channel.send(task2)
channel.close()

81. send(task2)
Producer-consumer solution
val channel = Channel<Task>()
processing task1
val task =
channel.receive()
processTask(task)
channel.send(task1)
channel.send(task2)
channel.close()

82. send(task2)
Producer-consumer solution
val channel = Channel<Task>()
waiting for “receive”
val task =
channel.receive()
processTask(task)
processing task1
channel.send(task1)
channel.send(task2)
channel.close()

83. send(task2)
Producer-consumer solution
val channel = Channel<Task>()
receiveRendezvous!
val task =
channel.receive()
processTask(task)
val task =
channel.receive()
processTask(task)
processing task1
channel.send(task1)
channel.send(task2)
channel.close()

84. Producer-consumer solution
val channel = Channel<Task>()
processing task2
val task =
channel.receive()
processTask(task)
val task =
channel.receive()
processTask(task)
processing task1
channel.send(task1)
channel.send(task2)
channel.close()

85. consumer #1
producer #1
send tasks
receive tasks
consumer #2
Producer-consumer solution: many tasks
val channel = Channel<Task>()

86. Producer-consumer solution: many tasks
val channel = Channel<Task>()
async {
for (i in 1..N) {
channel.send(Task( "task$i"))
}
channel.close()
}
producer

87. Producer-consumer solution: many tasks
val channel = Channel<Task>()
...
async { worker(channel) }
async { worker(channel) }
consumer #1
consumer #2
suspend fun worker(channel: Channel<Task>) {
for (task in channel) {
processTask(task)
}
}
calls receive while iterating

88. Flows

89. Flow
•suspend-based reactive stream
flow { emit(value) }
.map { transform(it) }
.filter { condition(it) }
.catch { exception -> log(exception) }
.collect { process(it) }

90. Integration with RxJava
Use extension functions:
•flow.asPublisher()
•publisher.asFlow()

91. Backpressure
•Backpressure happens automatically
thanks to suspension mechanism

92. More about coroutines

93. Store4 - Migrating a library from RxJava To Coroutines
Coroutines Case Study - Cleaning Up An Async API
Many successful stories

94. coroutines
async/await
Language
Library
channels
flows
kotlinx.coroutines
yield

95. Hands-on lab “Intro to coroutines & channels”
http://kotl.in/hands-on

96. •“Deep dive into coroutines on JVM” (KotlinConf 2017)
•“Kotlin Coroutines in Practice” (KotlinConf 2018)
•by Roman Elizarov

97. Have a nice Kotlin!
XJUIDPSPVUJOFT