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

Kotlin Coroutines

Kotlin Coroutines

An introduction to Kotlin coroutines

Wolfram Rittmeyer

March 27, 2019
Tweet

More Decks by Wolfram Rittmeyer

Other Decks in Programming

Transcript

  1. Why concurrency? • Support multiple clients (e.g. server) • React

    to user input while doing other tasks (UI) • Faster process results (fork/join) • And so on @RittmeyerW
  2. Concurrency != Threads • Concurrency possible without threads • See

    JS which runs (mostly) on one thread • For Java think of NIO / NIO.2 • Really needed only for heavy computation @RittmeyerW
  3. Concurrency approaches • Communication via shared state • Callbacks •

    Futures / Promises • Reactive Extensions • Coroutines @RittmeyerW
  4. Blocking solution private fun firstTask(): String { log("about to create

    the task...") Thread.sleep(500) log("task is created...") return "The answer to life, universe and everything" } @RittmeyerW
  5. Blocking solution fun runTasks() { log("blocking solution") val task =

    firstTask() val result = secondTask(task) log(result) } @RittmeyerW
  6. Solution with callbacks private fun firstTask(callback: (String) -> Unit) {

    log("about to create the task...") thread { Thread.sleep(500) log("task is created...") callback("The answer to life, universe and everything") } } @RittmeyerW
  7. Solution with callbacks fun runTasks() { log("using callbacks") firstTask {

    task -> secondTask(task) { result -> log(result) } } } @RittmeyerW
  8. Solution with futures private fun firstTask(): CompletableFuture<String> { log("about to

    create the task...") val future = CompletableFuture<String>() thread { Thread.sleep(500) log("task is created...") future.complete("The answer to life, universe and everything") } return future } @RittmeyerW
  9. Solution with futures fun runTasks() { firstTask() .thenCompose { task

    -> secondTask(task) } .thenAccept { result -> log(result) } } @RittmeyerW
  10. Solution with coroutines private suspend fun firstTask(): String { log("about

    to create the task...") delay(500) log("task is created...") return "The answer to life, universe and everything" } @RittmeyerW
  11. Solution with coroutines fun runTasks() = runBlocking { launch {

    val firstResult = firstTask() val secondResult = secondTask(firstResult) log(secondResult) } } @RittmeyerW
  12. Suspending Functions • Special modifier suspend • Contains suspension points

    • That’s where execution pauses • And resumes after the work is done – Details later on :-) @RittmeyerW
  13. Suspending Functions suspend fun createTask(): String { log("about to create

    the task...") delay(500) log("task is created...") return "The answer to life, universe and everything" } @RittmeyerW
  14. Suspending Functions suspend fun createTask(): String { log("about to create

    the task...") delay(500) log("task is created...") return "The answer to life, universe and everything" } @RittmeyerW
  15. Suspending Functions fun createTask(): String { log("about to create the

    task...") delay(500) log("task is created...") return "The answer to life, universe and everything" } @RittmeyerW
  16. Coroutine Builders • Creates coroutine • Needed to call suspending

    functions • Expects a suspending lambda • Standard builders: – runBlocking, launch and async @RittmeyerW
  17. Coroutine Builders - runBlocking fun main() { runBlocking { //

    do some stuff suspendingly: delay(500) println("after delay") // do something else; suspendingly or not } println("after runBlocking will be printed") } @RittmeyerW
  18. Coroutine Builders - launch val job = launch { val

    firstResult = firstTask() val secondResult = secondTask(firstResult) log(secondResult) } // job.join() @RittmeyerW
  19. Coroutine Builders - async suspend fun asyncSample() = coroutineScope {

    val deferred1 = async { someTaskAsync() } val deferred2 = async { someOtherTaskAsync() } deferred1 == deferred2 } @RittmeyerW
  20. Coroutine Scopes • Builders are extensions of it – With

    the exception of runBlocking • Define the scope for coroutines @RittmeyerW
  21. Coroutine Scopes • Should match lifecycle of your objects –

    On Android: Activity/Fragment – On Backend: One request • GlobalScope should be used sparingly @RittmeyerW
  22. Coroutine Scopes class SomeActivity : AppCompatActivity(), CoroutineScope { lateinit var

    job: Job override val coroutineContext: CoroutineContext get() = Dispatchers.Main + job override fun onCreate(savedInstanceState: Bundle?) { super.onCreate(savedInstanceState) job = Job() } // … onDestroy() on next slide } @RittmeyerW
  23. Coroutine Scopes class SomeActivity : AppCompatActivity() { private val job

    = Job() private var uiScope = CoroutineScope(job + Dispatchers.Main) override fun onCreate(savedInstanceState: Bundle?) { super.onCreate(savedInstanceState) uiScope.launch { // call suspending functions } } // ... onDestroy() on next slide } @RittmeyerW
  24. Coroutine Scopes class MainActivity : AppCompatActivity() { private val job

    = Job() private var uiScope = CoroutineScope(job + Dispatchers.Main) // ... onCreate() on previous slide override fun onDestroy() { super.onDestroy() job.cancel() } } @RittmeyerW
  25. Jobs • Provide access to state of coroutine • Provide

    possibility to cancel • Arranged in parent child hierarchies • Can be joined to other jobs @RittmeyerW
  26. Dispatcher • Define the threading strategy • For most use

    cases predefined ones: – Default – Main / Main.immediate – Unconfined – IO @RittmeyerW
  27. Dispatcher • Use with builders – launch(Dispatchers.IO) { … }

    • Switch when needed – withContext(Dispatchers.IO) { … } • newSingleThreadContext("specialThread") @RittmeyerW
  28. Cancelling coroutines • Cancelling one job cancels all its childrens

    • For time based cancellation use – withTimeout(delayInMs) – withTimeoutOrNull(delayInMs) @RittmeyerW
  29. Cooperative Cancellation • Suspending functions have to support cancellation •

    Standard library functions support it • But others? Your own? – Be careful! @RittmeyerW
  30. Depends on scope • If needed: Use different scope •

    Consider using SupervisorJob – Children fail independently @RittmeyerW
  31. ExceptionHandlers val exceptionHandler: CoroutineExceptionHandler = CoroutineExceptionHandler { ctx, exception ->

    println("exception in handler: $exception - $ctx") exception.printStackTrace() } // ... val scope = CoroutineScope( Dispatchers.Unconfined + job + exceptionHandler ) @RittmeyerW
  32. Replace Dispatchers • Inject dispatchers (IO / Main) • Replace

    them in tests with Unconfined @RittmeyerW
  33. Use KotlinTest • Spec methods are suspending • Thus no

    need for runBlocking • Or GlobalScope.async { … } @RittmeyerW
  34. And / Or use mockk • Great support for mocking

    with coroutines • coEvery { definition } returns someObject • coVerify someObject shouldBe expected @RittmeyerW
  35. Coroutines and Java • Compiler cannot manipulate Java code •

    Thus no direct coroutines usage from Java • Interoperability of shared concepts @RittmeyerW
  36. Coroutines and futures • Integrates with CompletableFuture • Only JDK8

    API (Android: SDK 24) – CompletionStage.await() – CompletionStage.asDeferred() – Deferred.asCompletableFuture() – future CoroutineBuilder @RittmeyerW
  37. Use futures within coroutines • Easily convert existing futures val

    res = someFuture().await() log(res) @RittmeyerW
  38. Convert coroutines to futures • Easily create a future fun

    asFuture(): CompletableFuture<Int> = future { val firstResult = firstTask() val secondResult = secondTask(firstResult) secondResult } @RittmeyerW
  39. Wrap a custom API fun startApiWithCallback(): Deferred<Result> { val deferred

    = CompletableDeferred<Result>() someApiWithCallback { result -> if (deferred.isActive) { deferred.complete(result) } } return deferred } @RittmeyerW
  40. Wrap a custom API suspend fun startApiWithCallback(): String = suspendCancellableCoroutine

    { continuation -> continuation.invokeOnCancellation { // cancel api call... } println("starting with continuation...") callApiWithContinuation(continuation) } @RittmeyerW
  41. Wrap a custom API private fun callApiWithContinuation( continuation: CancellableContinuation<String>) {

    someApiWithCallback { result -> if (continuation.isActive) { when (result) { is Result.Success -> continuation.resume(result.success) is Result.Error -> continuation.resumeWithException(result.error) } } } } @RittmeyerW
  42. Continuations • Expect a result after the suspension ends •

    Are heavily used internally @RittmeyerW
  43. Continuations public interface Continuation<in T> { public val context: CoroutineContext

    public fun resumeWith(result: Result<T>) } public inline fun <T> Continuation<T>.resume(value: T): Unit = resumeWith(Result.success(value)) public inline fun <T> Continuation<T>.resumeWithException(exception: Throwable): Unit = resumeWith(Result.failure(exception)) @RittmeyerW
  44. Continuations public interface Continuation<in T> { public val context: CoroutineContext

    public fun resumeWith(result: Result<T>) } public inline fun <T> Continuation<T>.resume(value: T): Unit = resumeWith(Result.success(value)) public inline fun <T> Continuation<T>.resumeWithException(exception: Throwable): Unit = resumeWith(Result.failure(exception)) @RittmeyerW
  45. Injection of function argument • Compiler modifies signature: – Additional

    function argument • of type Continuation – Different return type • It’s now always Any (Object) @RittmeyerW
  46. Injection of function argument @RittmeyerW public final Object sampleFunction(@NotNull Continuation

    var1) { println("do something") delay(1_000) // suspension point println("done with something") }
  47. Sample code continued @RittmeyerW public final Object sampleFunction(@NotNull Continuation var1)

    { println("do something") delay(1_000) // suspension point println("done with something") }
  48. Continuation object of sample @RittmeyerW $continuation = new ContinuationImpl(var1) {

    Object result; int label; Object L$0; @Nullable public final Object invokeSuspend(@NotNull Object result) { this.result = result; this.label |= Integer.MIN_VALUE; return SourceSampleSuspendingFunction.this.sampleFunction(this); } };
  49. Simplified sample code @RittmeyerW fun sampleFunction(continuation: Continuation<Int>) : Any {

    val myContinuation = continuation as $continuation when (myContinuation.label) { // ... } println("done with something") return Unit }
  50. Simplified sample code @RittmeyerW when (myContinuation.label) { 0 -> {

    println("do something") val res = delay(1_000, continuation) if (res == COROUTINE_SUSPENDED) { myContinuation.label = 1 return COROUTINE_SUSPENDED } } 1 -> { // error handling omitted for brevity // proper return values would be stored } else -> throw IllegalStateException("something went wrong") }
  51. Cooperative multitasking • Mostly implemented by the compiler • Coroutines

    use cooperative multitasking • Be careful with blocking threads @RittmeyerW