Kotlin Coroutines

1. Kotlin Coroutines Mario Arias - Cake Solutions Ltd, Lunch & Learn March 2018 @dh44t
2. Topics About me A tiny introduction to Kotlin Function types Extension points Function composition, currying and partial application funKTionale’s Option, Either and Disjunction Future developments
3. Software Engineer at Cake Solutions 12+ years of experience with JVM technologies Spring certified trainer 7+ years with Scala 5+ years with Kotlin funKTionale Arrow (funTKtionale + Kategory) RxKotlin original developer and former team leader Spring JDBC Kotlin support
4. Co-author of Functional Kotlin March 2018
5. What is Kotlin? Statically typed programming language for modern multiplatform applications
6. Hello world package org.cakesolutions.kotlin fun main(args: Array<String>) { println("Hello Reddish!") }
7. Features •Classes •sealed (ADT), data •Interfaces with body methods •Objects •Companion objects •Enum •Generics •Named and default parameters
8. Features II •Control structures •if else, when •do, while, for •Expressions •Null Safety •High order functions •Delegates •Extension functions
9. Null Safety package org.cakesolutions.kotlin val hello0: String = "" val hello1: String = null //Null can not be a value of a non-null type String val hello2: String? = null val reversed0: String? = hello2?.reversed() val reversed1: String = reversed0.reversed()
10. Null Safety II Any String Nothing
11. Null Safety III Any String Nothing Any? String?
12. Functions /** A function that takes 1 argument. */ public interface Function1<in P1, out R> : Function<R> { /** Invokes the function with the specified argument. */ public operator fun invoke(p1: P1): R }
13. Functions II fun doWithTwoNumbers(a: Int, b: Int, f: (Int, Int)  Int): Int { return f(a, b) } fun main(args: Array<String>) { val x = doWithTwoNumbers(2, 3, { a, b  a + b }) doWithTwoNumbers(x, 3) { a, b  a * b } }
14. Functions III Create your own control structures!! fun unless(conditional: Boolean, body: ()  Unit) { if (!conditional) { body() } } fun main(args: Array<String>) { val age: Int = //Magic here! unless(age > 18) { //Play video games all day long } }
15. Delegated property lazy is NOT a language feature but a library function val lazyValue: String by lazy { println("computed") "Hello" } fun main(args: Array<String>) { println(lazyValue) println(lazyValue) }
16. Delegated property import kotlin.properties.Delegates class User { var name: String by Delegates.observable("<No name>") { property, oldValue, newValue  println("$oldValue  $newValue") } } fun main(args: Array<String>) { val user = User() user.name = "first" user.name = "second" }
17. Extension functions fun String.fromReddish() = "$this, FROM Reddish" fun main(args: Array<String>) { println("Hello World".fromReddish()) }
18. Extension functions II operator fun String.invoke(suffix: String) = "$this$suffix" fun main(args: Array<String>) { println("This isn't a function"("")) }
19. Extension functions III fun main(args: Array<String>) { val myList = listOf(1, 2, 3, 4) //call to method reference .map(InttoString) .map { i  i + i } .map(StringtoInt) .filter { it > 20 } // `it` is an implicit parameter println(myList) }
20. Combine all the things!!! one-line unless from Ruby!! infix fun (()  Unit).unless(condition: Boolean) { if (!condition) { this() } } fun main(args: Array<String>) { val age: Int = //Magic { /*pay taxes*/ } unless (age < 18) }
21. Inline fun <T> time(body: ()  T): Pair<T, Long> { val startTime = System.currentTimeMillis() val v = body() val endTime = System.currentTimeMillis() return v to endTime - startTime } inline fun <T> inTime(body: ()  T): Pair<T, Long> { val startTime = System.currentTimeMillis() val v = body() val endTime = System.currentTimeMillis() return v to endTime - startTime }
22. Inline II fun main(args: Array<String>) { val b = time { Thread.sleep(1000) "boring" } val i = inTime { Thread.sleep(1000) "inline" } println(b) println(i) }
23. Inline III @NotNull public static final <T> Pair<T, Long> time(@NotNull final Function0<? extends T> body) { Intrinsics.checkParameterIsNotNull((Object)body, "body"); final long startTime = System.currentTimeMillis(); final Object v = body.invoke(); final long endTime = System.currentTimeMillis(); return (Pair<T, Long>)TuplesKt.to(v, (Object)(endTime - startTime)); } @NotNull public static final <T> Pair<T, Long> inTime(@NotNull final Function0<? extends T> body) { Intrinsics.checkParameterIsNotNull((Object)body, "body"); final long startTime = System.currentTimeMillis(); final Object v = body.invoke(); final long endTime = System.currentTimeMillis(); return (Pair<T, Long>)TuplesKt.to(v, (Object)(endTime - startTime)); } Decompiled bytecode
24. Inline IV public static final void main(@NotNull final String[] args) { Intrinsics.checkParameterIsNotNull((Object)args, "args"); final Pair b = time( (kotlin.jvm.functions.Function0<?>)_05_inlineKt$main$b._05_inlineKt$main$b$1.INSTANCE ); final long startTime$iv = System.currentTimeMillis(); Thread.sleep(1000L); final Object v$iv = "inline"; final long endTime$iv = System.currentTimeMillis(); final Pair i = TuplesKt.to(v$iv, (Object)(endTime$iv - startTime$iv)); System.out.println(b); System.out.println(i); }
25. Inline V static final class _05_inlineKt$main$b$1 extends Lambda implements Function0<String> { public static final _05_inlineKt$main$b$1 INSTANCE; @NotNull public final String invoke() { Thread.sleep(1000L); return "boring"; } static { _05_inlineKt$main$b$1.INSTANCE = new _05_inlineKt$main$b$1(); } }
26. Type-safe builders fun <T> logged(body: Logger.()  T): T { val log = LoggerFactory.getLogger("builder") return log.body() } fun main(args: Array<String>) { val i = logged { info("THIS is INFO") debug("THIS is DEBUG") error("THIS is ERROR") 1 + 2 } }
27. DSLs class FxApp : App(FxViewclass) class FxView : View() { override val root = vbox { label("Reddish App") button("Press me") } } fun main(args: Array<String>) { Application.launch(FxAppclass.java, *args) }
28. Coroutines but first, Threads import kotlin.concurrent.thread fun main(args: Array<String>) { val computation = thread { Thread.sleep(1000) println("World!") } print("Hello, ") computation.join() }
29. Hardware and JVM Threads a simplistic, and incomplete, explanation Core 1 T1 T2 T3 Core 2 T4 T.. TN
30. Monitoring utils VisualVM class MonitorUtil { private val scanner = Scanner(System.`in`) fun awaitEnter(){ println("Press enter to continue") scanner.nextLine() } } fun monitored(body: MonitorUtil.()  Unit) { val monitor = MonitorUtil() monitor.body() monitor.awaitEnter() }
31. 1000 Threads fun main(args: Array<String>) = monitored { awaitEnter() //VisualVM val threads = List(1000) { thread { Thread.sleep(1000) print('.') } } threads.forEach(Threadjoin) }
32. Executors fun main(args: Array<String>) = monitored { awaitEnter() val pool = Executors.newFixedThreadPool(128) repeat(10000) { pool.submit { Thread.sleep(1000) print('.') } } pool.shutdown() pool.awaitTermination(2, TimeUnit.SECONDS) }
33. Coroutines “..a coroutine is a very light thread that runs a block of code and has a similar life cycle, but can complete with a return value or an exception. Technically, a coroutine is an instance of a suspendable computation, a computation that may suspend. Coroutines aren't bound to a particular thread and can suspend in one Thread and resume execution in a different one” From the book Functional Kotlin
34. Coroutines II Core 1 T1 T2 T3 Core 2 T4 T.. TN Coroutines
35. Coroutines III import kotlinx.coroutines.experimental.delay import kotlinx.coroutines.experimental.launch import kotlinx.coroutines.experimental.runBlocking fun main(args: Array<String>) = runBlocking { val job = launch { delay(1000) println("World!") } print("Hello, ") job.join() }
36. Coroutines IV Concept Description Coroutine Light thread Suspending function A function with a suspend modifier. It can suspend a coroutine without blocking the thread Suspending lambda A lambda function with a suspend modifier Coroutine builder A function that takes a suspending lambda Suspension point A point where a suspending function is invoked Continuation The state of a suspended coroutine at a suspension point
37. Coroutines V fun main(args: Array<String>) = runBlocking { val job = launch { delay(1000) println("World!") } print("Hello, ") job.join() } Suspension point Suspension point Suspending function Coroutine builder Coroutine builder Continuation Suspending lambda Suspending lambda Suspending function
38. 10000 Coroutines fun main(args: Array<String>) = monitored { awaitEnter() runBlocking { val jobs = List(10000) { launch { delay(1000) print(".") } } jobs.forEach { it.join() } } }
39. Shared state suspend fun repeatInParallel(times: Int, block: suspend ()  Unit) { val job = launch { repeat(times) { launch(coroutineContext) { block() } } } job.join() }
40. Shared state II fun main(args: Array<String>) = runBlocking { var counter = 0 val time = measureTimeMillis { repeatInParallel(1_000_000) { counter } } println("counter = $counter") //EPIC FAIL println("time = $time") }
41. Atomic Integer import java.util.concurrent.atomic.AtomicInteger fun main(args: Array<String>) = runBlocking { val counter = AtomicInteger(0) val time = measureTimeMillis { repeatInParallel(1_000_000) { counter.incrementAndGet() } } println("counter = ${counter.get()}") println("time = $time") }
42. Single thread context fun main(args: Array<String>) = runBlocking { var counter = 0 val counterContext = newSingleThreadContext("CounterContext") val time = measureTimeMillis { repeatInParallel(1_000_000) { withContext(counterContext) { counter } } } println("counter = $counter") println("time = $time") }
43. Mutex fun main(args: Array<String>) = runBlocking { var counter = 0 val mutex = Mutex() val time = measureTimeMillis { repeatInParallel(1_000_000) { mutex.withLock { counter } } } println("counter = $counter") println("time = $time") }
44. Actors I sealed class CounterMsg object IncCounter : CounterMsg() class GetCounter(val response: CompletableDeferred<Int>) : CounterMsg() fun counterActor(start: Int) = actor<CounterMsg> { var counter = start for (msg in channel) { when (msg) { is IncCounter  counter is GetCounter  msg.response.complete(counter) } } }
45. Actors II fun main(args: Array<String>) = runBlocking { val counterActor = counterActor(0) val time = measureTimeMillis { repeatInParallel(1_000_000) { counterActor.send(IncCounter) } } val counter = CompletableDeferred<Int>() counterActor.send(GetCounter(counter)) println("counter = ${counter.await()}") println("time = $time") }
46. Other uses import arrow.core.Some fun main(args: Array<String>) { val a = Some(1) val b = Some(2) val c = a.flatMap { x  b.map { y  x + y } } println("c = $c") }
47. Monadic comprehension import arrow.core.Option import arrow.core.Some import arrow.core.ev import arrow.core.monad import arrow.typeclasses.binding fun main(args: Array<String>) { val a = Some(1) val b = Some(2) val c = Option.monad().binding { val x = a.bind() val y = b.bind() x + y }.ev() println("c = $c") } continuation: flatMap continuation: map