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

Kotlin 1.3 Webinar Coroutines Presentation

Kotlin 1.3 Webinar Coroutines Presentation

Kotlin Coroutines is a foundational feature in Kotlin language that enables easy-to-write/easy-to-read asynchronous programming and more. For those of you who are familiar with other asynchronous programming abstractions like callbacks, futures/promises, and async/await, we’ll give an overview of what are the similarities and differences, and highlight the key features of Kotlin’s design.

Roman Elizarov

December 05, 2018
Tweet

More Decks by Roman Elizarov

Other Decks in Programming

Transcript

  1. Callbacks: before fun requestToken(): Token { // makes request for

    a token & waits return token // returns result when received } 1
  2. Callbacks: after fun requestTokenAsync(cb: (Token) -> Unit) { // makes

    request for a token, invokes callback when done // returns immediately } 1 callback
  3. Callbacks: before fun requestTokenAsync(cb: (Token) -> Unit) { … }

    fun createPost(token: Token, item: Item): Post { // sends item to the server & waits return post // returns resulting post } 2
  4. Callbacks: after fun requestTokenAsync(cb: (Token) -> Unit) { … }

    fun createPostAsync(token: Token, item: Item, cb: (Post) -> Unit) { // sends item to the server, invokes callback when done // returns immediately } 2 callback
  5. Callbacks: before fun requestTokenAsync(cb: (Token) -> Unit) { … }

    fun createPostAsync(token: Token, item: Item, cb: (Post) -> Unit) { … } fun processPost(post: Post) { … } fun postItem(item: Item) { val token = requestToken() val post = createPost(token, item) processPost(post) }
  6. Callbacks: after fun requestTokenAsync(cb: (Token) -> Unit) { … }

    fun createPostAsync(token: Token, item: Item, cb: (Post) -> Unit) { … } fun processPost(post: Post) { … } fun postItem(item: Item) { requestTokenAsync { token -> createPostAsync(token, item) { post -> processPost(post) } } } aka “callback hell” This is simplified. Handling exceptions makes it a real mess
  7. Futures: before fun requestTokenAsync(cb: (Token) -> Unit) { // makes

    request for a token, invokes callback when done // returns immediately } 1
  8. Futures: after fun requestTokenAsync(): Promise<Token> { // makes request for

    a token // returns promise for a future result immediately } 1 future
  9. Futures: before fun requestTokenAsync(): Promise<Token> { … } fun createPostAsync(token:

    Token, item: Item, cb: (Post) -> Unit) { // sends item to the server, invokes callback when done // returns immediately } 2
  10. Futures: after fun requestTokenAsync(): Promise<Token> { … } fun createPostAsync(token:

    Token, item: Item): Promise<Post> { // sends item to the server // returns promise for a future result immediately } future 2
  11. Futures: before fun requestTokenAsync(): Promise<Token> { … } fun createPostAsync(token:

    Token, item: Item): Promise<Post> … fun processPost(post: Post) { … } fun postItem(item: Item) { requestTokenAsync { token -> createPostAsync(token, item) { post -> processPost(post) } } }
  12. Futures: after fun requestTokenAsync(): Promise<Token> { … } fun createPostAsync(token:

    Token, item: Item): Promise<Post> … fun processPost(post: Post) { … } fun postItem(item: Item) { requestTokenAsync() .thenCompose { token -> createPostAsync(token, item) } .thenAccept { post -> processPost(post) } }
  13. Futures: after fun requestTokenAsync(): Promise<Token> { … } fun createPostAsync(token:

    Token, item: Item): Promise<Post> … fun processPost(post: Post) { … } fun postItem(item: Item) { requestTokenAsync() .thenCompose { token -> createPostAsync(token, item) } .thenAccept { post -> processPost(post) } } Callbacks are still here
  14. Futures: after fun requestTokenAsync(): Promise<Token> { … } fun createPostAsync(token:

    Token, item: Item): Promise<Post> … fun processPost(post: Post) { … } fun postItem(item: Item) { requestTokenAsync() .thenCompose { token -> createPostAsync(token, item) } .thenAccept { post -> processPost(post) } } Composable & propagates exceptions No nesting indentation Callbacks are still here
  15. Futures: after fun requestTokenAsync(): Promise<Token> { … } fun createPostAsync(token:

    Token, item: Item): Promise<Post> … fun processPost(post: Post) { … } fun postItem(item: Item) { requestTokenAsync() .thenCompose { token -> createPostAsync(token, item) } .thenAccept { post -> processPost(post) } } But all those operators…
  16. Coroutines: before fun requestTokenAsync(): Promise<Token> { // makes request for

    a token // returns promise for a future result immediately } 1
  17. Coroutines: after suspend fun requestToken(): Token { // makes request

    for a token & suspends return token // returns result when received } 1 natural signature
  18. Coroutines: before suspend fun requestToken(): Token { … } fun

    createPostAsync(token: Token, item: Item): Promise<Post> { // sends item to the server // returns promise for a future result immediately } 2
  19. Coroutines: after suspend fun requestToken(): Token { … } suspend

    fun createPost(token: Token, item: Item): Post { // sends item to the server & suspends return post // returns result when received } 2 natural signature
  20. Coroutines: before suspend fun requestToken(): Token { … } suspend

    fun createPost(token: Token, item: Item): Post { … } fun processPost(post: Post) { … } fun postItem(item: Item) { requestTokenAsync() .thenCompose { token -> createPostAsync(token, item) } .thenAccept { post -> processPost(post) } }
  21. Coroutines: after suspend fun requestToken(): Token { … } suspend

    fun createPost(token: Token, item: Item): Post { … } fun processPost(post: Post) { … } suspend fun postItem(item: Item) { val token = requestToken() val post = createPost(token, item) processPost(post) }
  22. Coroutines: after suspend fun requestToken(): Token { … } suspend

    fun createPost(token: Token, item: Item): Post { … } fun processPost(post: Post) { … } suspend fun postItem(item: Item) { val token = requestToken() val post = createPost(token, item) processPost(post) } Like regular code
  23. Coroutines: after suspend fun requestToken(): Token { … } suspend

    fun createPost(token: Token, item: Item): Post { … } fun processPost(post: Post) { … } suspend fun postItem(item: Item) { val token = requestToken() val post = createPost(token, item) processPost(post) } suspension points
  24. • Custom higher-order functions Bonus features val post = retryIO

    { createPost(token, item) } Everything like in regular code
  25. Kotlin suspending functions callback Kotlin Java/JVM suspend fun createPost(token: Token,

    item: Item): Post { … } Object createPost(Token token, Item item, Continuation<Post> cont) { … }
  26. Kotlin suspending functions callback Kotlin Java/JVM Continuation is a generic

    callback interface suspend fun createPost(token: Token, item: Item): Post { … } Object createPost(Token token, Item item, Continuation<Post> cont) { … } interface Continuation<in T> { val context: CoroutineContext fun resumeWith(result: Result<T>) }
  27. Kotlin suspending functions callback Kotlin Java/JVM suspend fun createPost(token: Token,

    item: Item): Post { … } Object createPost(Token token, Item item, Continuation<Post> cont) { … } interface Continuation<in T> { val context: CoroutineContext fun resumeWith(result: Result<T>) }
  28. Kotlin suspending functions callback Kotlin Java/JVM suspend fun createPost(token: Token,

    item: Item): Post { … } Object createPost(Token token, Item item, Continuation<Post> cont) { … } interface Continuation<in T> { val context: CoroutineContext fun resumeWith(result: Result<T>) }
  29. interface Service { fun createPost(token: Token, item: Item): Call<Post> }

    suspend fun createPost(token: Token, item: Item): Post = serviceInstance.createPost(token, item).await() natural signature
  30. interface Service { fun createPost(token: Token, item: Item): Call<Post> }

    suspend fun createPost(token: Token, item: Item): Post = serviceInstance.createPost(token, item).await() Suspending extension function from integration library
  31. Callbacks everywhere suspend fun <T> Call<T>.await(): T { enqueue(object :

    Callback<T> { override fun onResponse(call: Call<T>, response: Response<T>) { // todo } override fun onFailure(call: Call<T>, t: Throwable) { // todo } }) }
  32. suspend fun <T> Call<T>.await(): T = suspendCoroutine { cont ->

    enqueue(object : Callback<T> { override fun onResponse(call: Call<T>, response: Response<T>) { if (response.isSuccessful) cont.resume(response.body()!!) else cont.resumeWithException(ErrorResponse(response)) } override fun onFailure(call: Call<T>, t: Throwable) { cont.resumeWithException(t) } }) }
  33. suspend fun <T> Call<T>.await(): T = suspendCoroutine { cont ->

    enqueue(object : Callback<T> { override fun onResponse(call: Call<T>, response: Response<T>) { if (response.isSuccessful) cont.resume(response.body()!!) else cont.resumeWithException(ErrorResponse(response)) } override fun onFailure(call: Call<T>, t: Throwable) { cont.resumeWithException(t) } }) }
  34. Install callback suspend fun <T> Call<T>.await(): T = suspendCoroutine {

    cont -> enqueue(object : Callback<T> { override fun onResponse(call: Call<T>, response: Response<T>) { if (response.isSuccessful) cont.resume(response.body()!!) else cont.resumeWithException(ErrorResponse(response)) } override fun onFailure(call: Call<T>, t: Throwable) { cont.resumeWithException(t) } }) }
  35. Install callback suspend fun <T> Call<T>.await(): T = suspendCoroutine {

    cont -> enqueue(object : Callback<T> { override fun onResponse(call: Call<T>, response: Response<T>) { if (response.isSuccessful) cont.resume(response.body()!!) else cont.resumeWithException(ErrorResponse(response)) } override fun onFailure(call: Call<T>, t: Throwable) { cont.resumeWithException(t) } }) }
  36. Analyze response suspend fun <T> Call<T>.await(): T = suspendCoroutine {

    cont -> enqueue(object : Callback<T> { override fun onResponse(call: Call<T>, response: Response<T>) { if (response.isSuccessful) cont.resume(response.body()!!) else cont.resumeWithException(ErrorResponse(response)) } override fun onFailure(call: Call<T>, t: Throwable) { cont.resumeWithException(t) } }) }
  37. Analyze response suspend fun <T> Call<T>.await(): T = suspendCoroutine {

    cont -> enqueue(object : Callback<T> { override fun onResponse(call: Call<T>, response: Response<T>) { if (response.isSuccessful) cont.resume(response.body()!!) else cont.resumeWithException(ErrorResponse(response)) } override fun onFailure(call: Call<T>, t: Throwable) { cont.resumeWithException(t) } }) } That’s all
  38. Coroutines revisited suspend fun requestToken(): Token { … } suspend

    fun createPost(token: Token, item: Item): Post { … } fun processPost(post: Post) { … } suspend fun postItem(item: Item) { val token = requestToken() val post = createPost(token, item) processPost(post) }
  39. Coroutines revisited suspend fun requestToken(): Token { … } suspend

    fun createPost(token: Token, item: Item): Post { … } fun processPost(post: Post) { … } fun postItem(item: Item) { val token = requestToken() val post = createPost(token, item) processPost(post) }
  40. Coroutines revisited suspend fun requestToken(): Token { … } suspend

    fun createPost(token: Token, item: Item): Post { … } fun processPost(post: Post) { … } fun postItem(item: Item) { val token = requestToken() val post = createPost(token, item) processPost(post) } Error: Suspend function 'requestToken' should be called only from a coroutine or another suspend function
  41. Coroutines revisited suspend fun requestToken(): Token { … } suspend

    fun createPost(token: Token, item: Item): Post { … } fun processPost(post: Post) { … } fun postItem(item: Item) { val token = requestToken() val post = createPost(token, item) processPost(post) } Can suspend execution
  42. Coroutines revisited suspend fun requestToken(): Token { … } suspend

    fun createPost(token: Token, item: Item): Post { … } fun processPost(post: Post) { … } fun postItem(item: Item) { val token = requestToken() val post = createPost(token, item) processPost(post) } Can suspend execution A regular function cannot
  43. Coroutines revisited suspend fun requestToken(): Token { … } suspend

    fun createPost(token: Token, item: Item): Post { … } fun processPost(post: Post) { … } fun postItem(item: Item) { val token = requestToken() val post = createPost(token, item) processPost(post) } Can suspend execution A regular function cannot One cannot simply invoke a suspending function
  44. Launch fun postItem(item: Item) { GlobalScope.launch { val token =

    requestToken() val post = createPost(token, item) processPost(post) } } coroutine builder
  45. fun postItem(item: Item) { GlobalScope.launch { val token = requestToken()

    val post = createPost(token, item) processPost(post) } } Fire and forget!
  46. fun postItem(item: Item) { GlobalScope.launch { val token = requestToken()

    val post = createPost(token, item) processPost(post) } } Fire and forget! We launch for its side effects
  47. fun postItem(item: Item) { GlobalScope.launch { val token = requestToken()

    val post = createPost(token, item) processPost(post) } } Returns immediately, coroutine works in background thread pool
  48. fun postItem(item: Item) { GlobalScope.launch { val token = requestToken()

    val post = createPost(token, item) processPost(post) } }
  49. fun postItem(item: Item) { GlobalScope.launch(Dispatchers.Main) { val token = requestToken()

    val post = createPost(token, item) processPost(post) } } Dispatcher Just specify the dispatcher
  50. fun postItem(item: Item) { GlobalScope.launch(Dispatchers.Main) { val token = requestToken()

    val post = createPost(token, item) processPost(post) } } Dispatcher And it gets executed on the Main thread
  51. Kotlin-way suspend fun postItem(item: Item) { val token = requestToken()

    val post = createPost(token, item) processPost(post) } Kotlin suspend fun requestToken(): Token { … } suspend fun createPost(token: Token, item: Item): Post { … } fun processPost(post: Post) { … }
  52. async Task postItem(Item item) { var token = await requestToken();

    var post = await createPost(token, item); processPost(post); } Classic-way C# approach to the same problem (also Python, TS, Dart, coming to JS) C# async Task<Token> requestToken() { … } async Task<Post> createPost(Token token, Item item) { … } void processPost(Post post) { … }
  53. async Task postItem(Item item) { var token = await requestToken();

    var post = await createPost(token, item); processPost(post); } Classic-way mark with async C# async Task<Token> requestToken() { … } async Task<Post> createPost(Token token, Item item) { … } void processPost(Post post) { … }
  54. async Task postItem(Item item) { var token = await requestToken();

    var post = await createPost(token, item); processPost(post); } Classic-way use await to suspend C# async Task<Token> requestToken() { … } async Task<Post> createPost(Token token, Item item) { … } void processPost(Post post) { … }
  55. async Task postItem(Item item) { var token = await requestToken();

    var post = await createPost(token, item); processPost(post); } Classic-way C# returns a future async Task<Token> requestToken() { … } async Task<Post> createPost(Token token, Item item) { … } void processPost(Post post) { … }
  56. Why no await keyword in Kotlin? The problem with async

    requestToken() VALID –> produces Task<Token> await requestToken() VALID –> produces Token concurrent behavior sequential behavior C# C# default
  57. Kotlin suspending functions are designed to imitate sequential behavior by

    default Concurrency is hard Concurrency has to be explicit
  58. Use-case for async var promise1 = loadImageAsync(name1); var promise2 =

    loadImageAsync(name2); async Task<Image> loadImageAsync(String name) { … } Start multiple operations concurrently C#
  59. Use-case for async var promise1 = loadImageAsync(name1); var promise2 =

    loadImageAsync(name2); var image1 = await promise1; var image2 = await promise2; async Task<Image> loadImageAsync(String name) { … } and then wait for them C#
  60. Use-case for async var result = combineImages(image1, image2); C# var

    promise1 = loadImageAsync(name1); var promise2 = loadImageAsync(name2); var image1 = await promise1; var image2 = await promise2; async Task<Image> loadImageAsync(String name) { … }
  61. fun loadImageAsync(name: String): Deferred<Image> = GlobalScope.async { … } val

    deferred1 = loadImageAsync(name1) val deferred2 = loadImageAsync(name2) Start multiple operations concurrently Kotlin
  62. fun loadImageAsync(name: String): Deferred<Image> = GlobalScope.async { … } val

    deferred1 = loadImageAsync(name1) val deferred2 = loadImageAsync(name2) val image1 = deferred1.await() val image2 = deferred2.await() and then wait for them await function Suspends until deferred is complete Kotlin
  63. fun loadImageAsync(name: String): Deferred<Image> = GlobalScope.async { … } val

    deferred1 = loadImageAsync(name1) val deferred2 = loadImageAsync(name2) val image1 = deferred1.await() val image2 = deferred2.await() val result = combineImages(image1, image2) Kotlin But what if this crashes? Crash?
  64. fun loadImageAsync(name: String): Deferred<Image> = GlobalScope.async { … } val

    deferred1 = loadImageAsync(name1) val deferred2 = loadImageAsync(name2) val image1 = deferred1.await() val image2 = deferred2.await() val result = combineImages(image1, image2) Kotlin Crash? This one leaks But what if this crashes?
  65. Idiomatic: Using async function when needed suspend fun loadImage(name: String):

    Image { … } Is defined as suspending function, not async
  66. Idiomatic: Using async function when needed suspend fun loadImage(name: String):

    Image { … } suspend fun loadAndCombine(name1: String, name2: String): Image = coroutineScope { val deferred1 = async { loadImage(name1) } val deferred2 = async { loadImage(name2) } combineImages(deferred1.await(), deferred2.await()) } Structured concurrency
  67. Idiomatic: Using async function when needed suspend fun loadImage(name: String):

    Image { … } suspend fun loadAndCombine(name1: String, name2: String): Image = coroutineScope { val deferred1 = async { loadImage(name1) } val deferred2 = async { loadImage(name2) } combineImages(deferred1.await(), deferred2.await()) }
  68. Idiomatic: Using async function when needed suspend fun loadImage(name: String):

    Image { … } suspend fun loadAndCombine(name1: String, name2: String): Image = coroutineScope { val deferred1 = async { loadImage(name1) } val deferred2 = async { loadImage(name2) } combineImages(deferred1.await(), deferred2.await()) }
  69. Idiomatic: Using async function when needed suspend fun loadImage(name: String):

    Image { … } suspend fun loadAndCombine(name1: String, name2: String): Image = coroutineScope { val deferred1 = async { loadImage(name1) } val deferred2 = async { loadImage(name2) } combineImages(deferred1.await(), deferred2.await()) }
  70. Kotlin approach to async requestToken() VALID –> produces Token async

    { requestToken() } VALID –> produces Deferred<Token> sequential behavior concurrent behavior Kotlin Kotlin default
  71. Structured concurrency suspend fun loadImage(name: String): Image { … }

    suspend fun loadAndCombine(name1: String, name2: String): Image = coroutineScope { val deferred1 = async { loadImage(name1) } val deferred2 = async { loadImage(name2) } combineImages(deferred1.await(), deferred2.await()) }
  72. Crash suspend fun loadImage(name: String): Image { … } suspend

    fun loadAndCombine(name1: String, name2: String): Image = coroutineScope { val deferred1 = async { loadImage(name1) } val deferred2 = async { loadImage(name2) } combineImages(deferred1.await(), deferred2.await()) } Crashes?
  73. Cancels scope suspend fun loadImage(name: String): Image { … }

    suspend fun loadAndCombine(name1: String, name2: String): Image = coroutineScope { val deferred1 = async { loadImage(name1) } val deferred2 = async { loadImage(name2) } combineImages(deferred1.await(), deferred2.await()) } Cancels
  74. Cancels children suspend fun loadImage(name: String): Image { … }

    suspend fun loadAndCombine(name1: String, name2: String): Image = coroutineScope { val deferred1 = async { loadImage(name1) } val deferred2 = async { loadImage(name2) } combineImages(deferred1.await(), deferred2.await()) } Cancels
  75. Fibonacci sequence val fibonacci = sequence { var cur =

    1 var next = 1 while (true) { yield(cur) val tmp = cur + next cur = next next = tmp } }
  76. Fibonacci sequence val fibonacci = sequence { var cur =

    1 var next = 1 while (true) { yield(cur) val tmp = cur + next cur = next next = tmp } } println(fibonacci.take(10).toList())
  77. Fibonacci sequence val fibonacci = sequence { var cur =

    1 var next = 1 while (true) { yield(cur) val tmp = cur + next cur = next next = tmp } } println(fibonacci.take(10).toList()) >> [1, 1, 2, 3, 5, 8, 13, 21, 34, 55]
  78. Fibonacci sequence val fibonacci = sequence { var cur =

    1 var next = 1 while (true) { yield(cur) val tmp = cur + next cur = next next = tmp } } A coroutine builder with restricted suspension
  79. Fibonacci sequence val fibonacci = sequence { var cur =

    1 var next = 1 while (true) { yield(cur) val tmp = cur + next cur = next next = tmp } } A suspending function in the scope of buildSequence
  80. Fibonacci sequence val fibonacci = sequence { var cur =

    1 var next = 1 while (true) { yield(cur) val tmp = cur + next cur = next next = tmp } } println(fibonacci.take(10).toList()) Synchronous