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

Server-Side Kotlin Coroutines

Server-Side Kotlin Coroutines

Kotlin was designed as a general-purpose programming language and with coroutines writing server-side code is a easy as never before on JVM. We'll dissect scalability and reliability problems of a traditional thread-based stacks and discuss the process of mitigating those issues by introducing asynchrony with Kotlin coroutines. We'll see how coroutine-based design naturally avoids common pitfalls of traditional asynchronous programming such as resource management, error handling and request cancellation, producing safe and reliable code using a concept of structured concurrency.

Roman Elizarov

April 30, 2019
Tweet

More Decks by Roman Elizarov

Other Decks in Programming

Transcript

  1. Speaker: Roman Elizarov • Professional developer since 2000 • Previously

    developed high-perf trading software @ Devexperts • Teach concurrent & distributed programming @ St. Petersburg ITMO University • Chief judge @ Northern Eurasia Contest / ICPC • Now team lead in Kotlin Libraries @ JetBrains elizarov @ relizarov
  2. ET 1 ET 2 ET N … Executor Threads DB

    Old-school client-server monolith Clients
  3. ET 1 ET 2 ET N … Executor Threads DB

    Incoming request Clients
  4. ET 1 ET 2 ET N … Executor Threads DB

    Sizing threads – easy Clients N = number of DB connections
  5. ET 1 ET 2 ET N … Executor Threads DB

    Old-school client-server monolith Clients
  6. ET 1 ET 2 ET N … Executor Threads DB

    Now with Services Service Clients
  7. ET 1 ET 2 ET N … Executor Threads Services

    everywhere … Service K Service 1 Service 2 Clients
  8. ET 1 ET 2 ET N … Executor Threads Sizing

    threads – not easy … N = ????? Service K Service 1 Service 2 Clients
  9. Complex business logic fun placeOrder(order: Order): Response { val account

    = accountService.loadAccout(order.accountId) … }
  10. Complex business logic fun placeOrder(order: Order): Response { val account

    = accountService.loadAccout(order.accountId) val margin = if (account.isOptionsAccount) … }
  11. Complex business logic fun placeOrder(order: Order): Response { val account

    = accountService.loadAccout(order.accountId) val margin = if (account.isOptionsAccount) { marginService.loadMargin(account) } else { defaultMargin } … }
  12. Complex business logic fun placeOrder(order: Order): Response { val account

    = accountService.loadAccout(order.accountId) val margin = if (account.isOptionsAccount) { marginService.loadMargin(account) } else { defaultMargin } return validateOrder(order, margin) }
  13. Complex business logic fun placeOrder(order: Order): Response { val account

    = accountService.loadAccout(order.accountId) val margin = if (account.isOptionsAccount) { marginService.loadMargin(account) } else { defaultMargin } return validateOrder(order, margin) }
  14. What if a service is slow? fun placeOrder(order: Order): Response

    { val account = accountService.loadAccout(order.accountId) val margin = if (account.isOptionsAccount) { marginService.loadMargin(account) } else { defaultMargin } return validateOrder(order, margin) } !
  15. ET 2 ET N … Executor Threads Clients Blocks threads

    … Service K Service 1 Service 2 ET 1! "
  16. ET 2 ET N … Executor Threads Clients Blocks threads

    … Service K Service 1 Service 2 ET 1! " !
  17. ET 2 ET N … Executor Threads Clients Blocks threads

    … Service K Service 1 Service 2 ET 1! " ! !
  18. ET 2 ET N … Executor Threads Clients Instead of

    blocking… … Service K Service 1 Service 2 ET 1! "
  19. ET 2 ET N … Executor Threads Release the thread

    … Service K Service 1 Service 2 ET 1 ! Clients
  20. Clients ET 2 ET N … Executor Threads Resume operation

    later … Service K Service 1 Service 2 ET 1 !
  21. Suspend behind the scenes suspend fun loadMargin(account: Account): Margin fun

    loadMargin(account: Account, cont: Continuation<Margin>) But why callback and not future?
  22. Performance! •Future is a synchronization primitive •Callback is a lower-level

    primitive •Integration with async IO libraries is easy
  23. ET 2 ET N … Executor Threads Clients Release thread?

    … Service K Service 1 Service 2 ET 1! "
  24. Server integrated with coroutines suspend fun placeOrder(order: Order): Response {

    // response from placed order cache return response }
  25. Server not integrated with coroutines fun placeOrder(order: Order) = GlobalScope.mono

    { // response from placed order cache return@mono response } Coroutine builder
  26. Suspend suspend fun placeOrder(order: Order): Response { val account =

    accountService.loadAccout(order.accountId) val margin = if (account.isOptionsAccount) { marginService.loadMargin(account) } else { defaultMargin } return validateOrder(order, margin) }
  27. Suspend suspend fun placeOrder(order: Order): Response { val account =

    accountService.loadAccout(order.accountId) val margin = if (account.isOptionsAccount) { marginService.loadMargin(account) } else { defaultMargin } return validateOrder(order, margin) } Invoke suspending funs
  28. Suspend is convenient suspend fun placeOrder(order: Order): Response { val

    account = accountService.loadAccout(order.accountId) val margin = if (account.isOptionsAccount) { marginService.loadMargin(account) } else { defaultMargin } return validateOrder(order, margin) } Invoke suspending funs Write regular code!
  29. Suspend is efficient suspend fun placeOrder(order: Order): Response { val

    account = accountService.loadAccount(order.accountId) val margin = marginService.loadMargin(account) return validateOrder(order, margin) } One object allocated
  30. Futures/Promises/Reactive – less efficient fun placeOrder(order: Order): Mono<Response> = accountService.loadAccountAsync(order.accountId)

    .flatMap { account -> marginService.loadMargin(account) } .map { margin -> validateOrder(order, margin) } Lambda allocated* Future allocated Lambda allocated Future allocated
  31. Let’s go deeper fun placeOrder(params: Params): Mono<Response> { // check

    pre-conditions return actuallyPlaceOrder(order) } fun actuallyPlaceOrder(order: Order): Mono<Response>
  32. Let’s go deeper (with coroutines) suspend fun placeOrder(params: Params): Response

    { // check pre-conditions return actuallyPlaceOrder(order) } suspend fun actuallyPlaceOrder(params: Params): Response Tail call optimization Tail call
  33. ET 2 ET N … Executor Threads Clients Thread pools

    ET 1 Service 1 Threads ST 2 ST M1 … S1 1 N = number of CPU cores M 1 = depends
  34. IO-bound withContext suspend fun loadAccount(order: Order): Account = withContext(dispatcher) {

    // some blocking code here.... } val dispatcher = Executors.newFixedThreadPool(M2).asCoroutineDispatcher()
  35. CPU-bound code suspend fun validateOrder(order: Order, margin: Margin): Response =

    withContext(compute) { // perform CPU-consuming computation } val compute = Executors.newFixedThreadPool(M3).asCoroutineDispatcher()
  36. ET 2 ET N … Executor Threads Clients Fine-grained control

    and encapsulation ET 1 Service 1 Threads S1 1 ST M1 Service 2 Threads S1 1 ST M 2 Service 3 Threads S1 1 ST M 3 Async IO-bound CPU-bound Never blocked
  37. withTimeout propagation suspend fun placeOrder(order: Order): Response = withTimeout(1000) {

    // code before loadMargin(account) // code after } suspend fun loadMargin(account: Account): Margin = suspendCoroutine { cont -> // install callback & use cont to resume }
  38. withTimeout propagation suspend fun placeOrder(order: Order): Response = withTimeout(1000) {

    // code before loadMargin(account) // code after } suspend fun loadMargin(account: Account): Margin = suspendCancellableCoroutine { cont -> // install callback & use cont to resume }
  39. withTimeout propagation suspend fun placeOrder(order: Order): Response = withTimeout(1000) {

    // code before loadMargin(account) // code after } suspend fun loadMargin(account: Account): Margin = suspendCancellableCoroutine { cont -> // install callback & use cont to resume cont.invokeOnCancellation { … } }
  40. Example fun placeOrder(order: Order): Response { val account = accountService.loadAccount(order)

    val margin = marginService.loadMargin(order) return validateOrder(order, account, margin) }
  41. Example fun placeOrder(order: Order): Response { val account = accountService.loadAccount(order)

    val margin = marginService.loadMargin(order) return validateOrder(order, account, margin) } No data dependencies
  42. Concurrency with async (futures) fun placeOrder(order: Order): Response { val

    account = accountService.loadAccountAsync(order) val margin = marginService.loadMarginAsync(order) return validateOrder(order, account.await(), margin.await()) }
  43. Concurrency with async (futures) fun placeOrder(order: Order): Response { val

    account = accountService.loadAccountAsync(order) val margin = marginService.loadMarginAsync(order) return validateOrder(order, account.await(), margin.await()) }
  44. Concurrency with async (futures) fun placeOrder(order: Order): Response { val

    account = accountService.loadAccountAsync(order) val margin = marginService.loadMarginAsync(order) return validateOrder(order, account.await(), margin.await()) } Fails?
  45. Concurrency with async (futures) fun placeOrder(order: Order): Response { val

    account = accountService.loadAccountAsync(order) val margin = marginService.loadMarginAsync(order) return validateOrder(order, account.await(), margin.await()) } Fails? Leaks!
  46. Concurrency with coroutines suspend fun placeOrder(order: Order): Response = coroutineScope

    { val account = async { accountService.loadAccount(order) } val margin = async { marginService.loadMargin(order) } validateOrder(order, account.await(), margin.await()) }
  47. Concurrency with coroutines suspend fun placeOrder(order: Order): Response = coroutineScope

    { val account = async { accountService.loadAccount(order) } val margin = async { marginService.loadMargin(order) } validateOrder(order, account.await(), margin.await()) }
  48. Concurrency with coroutines suspend fun placeOrder(order: Order): Response = coroutineScope

    { val account = async { accountService.loadAccount(order) } val margin = async { marginService.loadMargin(order) } validateOrder(order, account.await(), margin.await()) }
  49. Concurrency with coroutines suspend fun placeOrder(order: Order): Response = coroutineScope

    { val account = async { accountService.loadAccount(order) } val margin = async { marginService.loadMargin(order) } validateOrder(order, account.await(), margin.await()) } Fails?
  50. Concurrency with coroutines suspend fun placeOrder(order: Order): Response = coroutineScope

    { val account = async { accountService.loadAccount(order) } val margin = async { marginService.loadMargin(order) } validateOrder(order, account.await(), margin.await()) } Fails? Cancels
  51. Concurrency with coroutines suspend fun placeOrder(order: Order): Response = coroutineScope

    { val account = async { accountService.loadAccount(order) } val margin = async { marginService.loadMargin(order) } validateOrder(order, account.await(), margin.await()) } Fails? Cancels Cancels
  52. Concurrency with coroutines suspend fun placeOrder(order: Order): Response = coroutineScope

    { val account = async { accountService.loadAccount(order) } val margin = async { marginService.loadMargin(order) } validateOrder(order, account.await(), margin.await()) } Waits for completion of all children
  53. Without coroutine scope? suspend fun placeOrder(order: Order): Response { val

    account = async { accountService.loadAccount(order) } val margin = async { marginService.loadMargin(order) } return validateOrder(order, account.await(), margin.await()) }
  54. Without coroutine scope? suspend fun placeOrder(order: Order): Response { val

    account = async { accountService.loadAccount(order) } val margin = async { marginService.loadMargin(order) } return validateOrder(order, account.await(), margin.await()) } ERROR: Unresolved reference.
  55. Extensions of CoroutineScope fun <T> CoroutineScope.async( context: CoroutineContext = EmptyCoroutineContext,

    start: CoroutineStart = CoroutineStart.DEFAULT, block: suspend CoroutineScope.() -> T ): Deferred<T>
  56. Types as documentation fun foo(params: Params): Response suspend fun foo(params:

    Params): Response fun CoroutineScope.foo(params: Params): Response Fast, local Remote, or slow Side effect - bg
  57. Types are enforced fun foo(params: Params): Response suspend fun foo(params:

    Params): Response fun CoroutineScope.foo(params: Params): Response Not allowed But must provide scope explicitly Using coroutineScope { … } Fast, local Remote, or slow Side effect - bg
  58. Green threads / Fibers ET 2 ET N … Executor

    Threads ET 1 F 2 F M … Fibers F 1 ~ Coroutines Hidden from developer
  59. ET 2 ET N … Executor Threads Clients Threads ET

    1 Service 1 Threads S1 1 ST M1 Service 2 Threads S1 1 ST M 2 Service 3 Threads S1 1 ST M 3
  60. ET 2 ET N … Executor Threads Clients Solution –

    shared thread pool ET 1! ET N+1
  61. ET 2 ET N … Executor Threads Clients Solution –

    shared thread pool ET 1! ET N+1 ! ET N+2
  62. ET 2 ET N … Executor Threads Clients Solution –

    shared thread pool ET 1! ET N+1 ! … ET M! ET N+2 ET N+M
  63. withContext for IO suspend fun loadAccount(order: Order): Account = withContext(dispatcher)

    { // some blocking code here.... } val dispatcher = Executors.newFixedThreadPool(M2).asCoroutineDispatcher()
  64. withContext for Dispatсhers.IO suspend fun loadAccount(order: Order): Account = withContext(Dispatchers.IO)

    { // some blocking code here.... } No thread switch from Dispatchers.Default pool
  65. ET 2 ET N … Executor Threads Clients Solution –

    shared thread pool ET 1 Dispatchers.Default
  66. ET 2 ET N … Executor Threads Clients Solution –

    shared thread pool ET 1! ET N+1 ! … ET M! ET N+2 ET N+M Dispatchers.Default Dispatchers.IO
  67. Returning many responses suspend fun foo(params: Params): Response One response

    suspend fun foo(params: Params): List<Response> Many responses suspend fun foo(params: Params): ????<Response> Many responses async?
  68. Producer Builder fun CoroutineScope.foo(): ReceiveChannel<Int> = produce { for (i

    in 1..10) { send(i) delay(100) } } Channel type Can be async
  69. Consumer fun CoroutineScope.foo(): ReceiveChannel<Int> = produce { for (i in

    1..10) { send(i) delay(100) } } fun main() = runBlocking<Unit> { for (x in foo()) { println(x) } }
  70. Where’s the catch? fun CoroutineScope.foo(): ReceiveChannel<Int> = produce { for

    (i in 1..10) { send(i) delay(100) } } fun main() = runBlocking<Unit> { for (x in foo()) { println(x) } }
  71. Where’s the catch? fun CoroutineScope.foo(): ReceiveChannel<Int> = produce { for

    (i in 1..10) { send(i) delay(100) } } fun main() = runBlocking<Unit> { for (x in foo()) { println(x) } } Creates coroutine
  72. Try this! fun CoroutineScope.foo(): ReceiveChannel<Int> = produce { for (i

    in 1..10) { send(i) delay(100) } } fun main() = runBlocking<Unit> { foo() } Waits for completion of children !
  73. Flow example fun bar(): Flow<Int> = flow { for (i

    in 1..10) { emit(i) delay(100) } } ~ Asynchronous sequence
  74. Flow example fun bar(): Flow<Int> = flow { for (i

    in 1..10) { emit(i) delay(100) } } fun main() = runBlocking<Unit> { bar().collect { x -> println(x) } }
  75. Try this! fun bar(): Flow<Int> = flow { for (i

    in 1..10) { emit(i) delay(100) } } fun main() = runBlocking<Unit> { bar() } Flow is cold: describes the data, does not run it until collected !
  76. Flow example fun bar(): Flow<Int> = flow { for (i

    in 1..10) { emit(i) delay(100) } } fun main() = runBlocking<Unit> { bar() .map { it * it } .toList() } Write regular code! Similar to collections / sequences