Structured Concurrency

Roman Elizarov

July 11, 2019

Tweet Share

Transcript

1. Structured Concurrency Presented at Hydra Distributed Computing Conference, 2019 elizarov

   @ Roman Elizarov relizarov 1

2. 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

3. A short story of Kotlin Coroutines design 3

4. Inspired by async/await async Task PostItem(Item item) { var token

   = await RequestToken(); var post = await CreatePost(token, item); ProcessPost(post); } C# async modifier Returns a future 1. awaits a future other async functions return future 4 2. awaits a future

5. Inspired by async/await async Task PostItem(Item item) { var token

   = await RequestToken(); var post = await CreatePost(token, item); ProcessPost(post); } C# 5

6. Inspired by async/await async Task PostItem(Item item) { var token

   = await RequestToken(); var post = await CreatePost(token, item); ProcessPost(post); } C# Suspension points! But what about generators with yield return? 6

7. Kotlin DSL: Initial prototype fun postItem(item: Item) = async {

   val token = await(requestToken()) val post = await(createPost(token, item)) processPost(post) } Kotlin Coroutine builder Coroutine scope 7 Regular function

8. Kotlin DSL: Initial prototype fun postItem(item: Item) = async {

   val token = await(requestToken()) val post = await(createPost(token, item)) processPost(post) } Kotlin Coroutine builder await function Coroutine scope Suspending functions 8 Regular function

9. Kotlin DSL: Initial prototype fun postItem(item: Item) = async {

   val token = await(requestToken()) val post = await(createPost(token, item)) processPost(post) } Kotlin 9

10. Kotlin DSL: Initial prototype fun postItem(item: Item) = async {

    val token = await(requestToken()) val post = await(createPost(token, item)) processPost(post) } Kotlin async / await à future 10

11. Kotlin DSL: Initial prototype fun postItem(item: Item) = generate {

    val token = yield(requestToken()) val post = yield(createPost(token, item)) processPost(post) } Kotlin async / await à future generate / yield à sequence 11

12. Suspending functions? fun postItem(item: Item) = async { val token

    = await(requestToken()) val post = await(createPost(token, item)) processPost(post) } Kotlin 12

13. Suspending functions everywhere! fun postItem(item: Item) = async { val

    token = requestToken() val post = createPost(token, item) processPost(post) } Kotlin Returns a future 13

14. Suspending functions everywhere! suspend fun postItem(item: Item) { val token

    = requestToken() val post = createPost(token, item) processPost(post) } Kotlin Suspending function modifier 14

15. Suspending functions everywhere! suspend fun postItem(item: Item) { val token

    = requestToken() val post = createPost(token, item) processPost(post) } Kotlin 15

16. Suspending functions everywhere! func postItem(item Item) { token := requestToken()

    post := createPost(token, item) processPost(post) } Go 16

17. Prototyping libraries DSL for concurrency 17

18. 18 https://tour.golang.org/concurrency/1

19. A Tour of Go Concurrency #1 func say(s string) {

    for i := 0; i < 5; i++ { time.Sleep(100 * time.Millisecond) fmt.Println(s) } } func main() { go say("world") say("hello") } Go https://tour.golang.org/concurrency/1 19

20. A Tour of Go Concurrency #1 func say(s string) {

    for i := 0; i < 5; i++ { time.Sleep(100 * time.Millisecond) fmt.Println(s) } } func main() { go say("world") say("hello") } Go 20 https://tour.golang.org/concurrency/1

21. A Tour of Go Concurrency #1 func say(s string) {

    for i := 0; i < 5; i++ { time.Sleep(100 * time.Millisecond) fmt.Println(s) } } func main() { go say("world") say("hello") } Go 21 https://tour.golang.org/concurrency/1

22. A DSL for Concurrency: Prototype suspend fun say(s: String) {

    for (i in 0..4) { delay(100) println(s) } } fun main() = mainBlocking { go { say("world") } say("hello") } Kotlin 22 Suspending function modifier Suspending function

23. A DSL for Concurrency: Prototype suspend fun say(s: String) {

    for (i in 0..4) { delay(100) println(s) } } fun main() = mainBlocking { go { say("world") } say("hello") } Kotlin 23 Suspending function modifier Suspending function Coroutine builder Another builder

24. A DSL for Concurrency: Now suspend fun say(s: String) {

    for (i in 0..4) { delay(100) println(s) } } fun main() = runBlocking { launch { say("world") } say("hello") } Kotlin 24

25. A Tour of Go Concurrency #5 func fibonacci(c, quit chan

    int) { x, y := 0, 1 for { select { case c <- x: x, y = y, x+y case <-quit: fmt.Println("quit") return } } } Go https://tour.golang.org/concurrency/5 25

26. A Tour of Go Concurrency #5 func fibonacci(c, quit chan

    int) { x, y := 0, 1 for { select { case c <- x: x, y = y, x+y case <-quit: fmt.Println("quit") return } } } Go https://tour.golang.org/concurrency/5 26

27. A Tour of Go Concurrency #5 func fibonacci(c, quit chan

    int) { x, y := 0, 1 for { select { case c <- x: x, y = y, x+y case <-quit: fmt.Println("quit") return } } } Go https://tour.golang.org/concurrency/5 27

28. A Tour of Go Concurrency #5 func fibonacci(c, quit chan

    int) { x, y := 0, 1 for { select { case c <- x: x, y = y, x+y case <-quit: fmt.Println("quit") return } } } Go https://tour.golang.org/concurrency/5 28

29. A Tour of Go Concurrency #5 func fibonacci(c, quit chan

    int) { x, y := 0, 1 for { select { case c <- x: x, y = y, x+y case <-quit: fmt.Println("quit") return } } } Go https://tour.golang.org/concurrency/5 29

30. A DSL for Concurrency: Prototype suspend fun fib(c: SendChannel<Int>, quit:

    ReceiveChannel<Int>) { var x = 0 var y = 1 whileSelect { c.onSend(x) { val next = x + y x = y y = next true // continue while loop } quit.onReceive { println("quit") false // break while loop } } } Kotlin 30

31. A DSL for Concurrency: Prototype suspend fun fib(c: SendChannel<Int>, quit:

    ReceiveChannel<Int>) { var x = 0 var y = 1 whileSelect { c.onSend(x) { val next = x + y x = y y = next true // continue while loop } quit.onReceive { println("quit") false // break while loop } } } Kotlin 31 Library types

32. A DSL for Concurrency: Prototype suspend fun fib(c: SendChannel<Int>, quit:

    ReceiveChannel<Int>) { var x = 0 var y = 1 whileSelect { c.onSend(x) { val next = x + y x = y y = next true // continue while loop } quit.onReceive { println("quit") false // break while loop } } } Kotlin 32 Library types Select DSL

33. What more to wish? Is concurrency support a solved problem?

    33

34. launchUI { try { // suspend while asynchronously making request

    val result = makeRequest() // display result in UI display(result) } catch (exception: Throwable) { // process exception in UI } } Thread-bound UI programming 34

35. launchUI { try { // suspend while asynchronously making request

    val result = makeRequest() // display result in UI display(result) } catch (exception: Throwable) { // process exception in UI } } Thread-bound UI programming 35

36. launchUI { try { // suspend while asynchronously making request

    val result = makeRequest() // display result in UI display(result) } catch (exception: Throwable) { // process exception in UI } } Thread-bound UI programming 36

37. launchUI { try { // suspend while asynchronously making request

    val result = makeRequest() // display result in UI display(result) } catch (exception: Throwable) { // process exception in UI } } Thread-bound UI programming 37

38. launch(UI) { try { // suspend while asynchronously making request

    val result = makeRequest() // display result in UI display(result) } catch (exception: Throwable) { // process exception in UI } } Thread-bound UI programming 38 Coroutine context

39. launch(UI) { try { // suspend while asynchronously making request

    val result = makeRequest() // display result in UI run { display(result) } } catch (exception: Throwable) { // process exception in UI } } Thread-bound UI programming 39 Higher-order function Coroutine context

40. await higher-order function 40

41. await higher-order function 41 await Run(() => { Display(result); });

    C# Lambda

42. await higher-order function 42 var task = Run(() => {

    Display(result); }); await task; C# 1. Function called first 2. Then await

43. await higher-order function 43 var task = Run(() => {

    Display(result); }); await task; C# 1. Function called first 2. Then await A call to regular function

44. await higher-order function 44 run { display(result) } Kotlin Context

    is passed along the suspend callstack A call to suspending function

45. Cancellation The stumbling block of concurrency design 45

46. A Tour of Go Concurrency func fibonacci(c, quit chan int)

    { x, y := 0, 1 for { select { case c <- x: x, y = y, x+y case <-quit: fmt.Println("quit") return } } } Go https://tour.golang.org/concurrency/5 46 A CancellationToken

47. A Tour of Go Concurrency func fibonacci(c, quit chan int)

    { x, y := 0, 1 for { select { case c <- x: x, y = y, x+y case <-quit: fmt.Println("quit") return } } } Go https://tour.golang.org/concurrency/5 47 A CancellationToken A boilerplate

48. 48 quit chan

49. Pervasive cancellation context? 49 type Context interface { // …

    } Go https://golang.org/pkg/context/

50. Pervasive cancellation context? 50 type Context interface { // …

    // Done returns a channel that's closed when work done on // behalf of this context should be canceled. … Done() <-chan struct{} } Go https://golang.org/pkg/context/ At Google, we require that Go programmers pass a Context parameter as the first argument to every function on the call path between incoming and outgoing requests.

51. Lifetime prototype 51

52. Lifetime prototype 52 interface Lifetime : CoroutineContext.Element { // …

    } Kotlin

53. Lifetime prototype 53 interface Lifetime : CoroutineContext.Element { fun cancel(reason:

    Throwable? = null): Boolean } Kotlin

54. Lifetime prototype 54 interface Lifetime : CoroutineContext.Element { fun cancel(reason:

    Throwable? = null): Boolean fun onCompletion(handler: CompletionHandler): Registration } Kotlin

55. Job prototype 55 interface Job : CoroutineContext.Element { fun cancel(reason:

    Throwable? = null): Boolean fun onCompletion(handler: CompletionHandler): Registration } Kotlin

56. Job prototype: explicit cancel 56 val job = launch {

    say("world") } Kotlin

57. Job prototype: explicit cancel 57 val job = launch {

    say("world") } job.cancel() Kotlin

58. Job prototype: explicit cancel 58 val job = launch {

    say("world") } job.cancel() Kotlin delay(…) throw CancellationException()

59. Job prototype: higher-order operators 59 withTimeout(duration) { doSomething() } Kotlin

    job1 job2 Prototype worked like a charm

60. Children coroutines Nesting concurrent computations 60

61. Concurrent decomposition 61 val job1 = launch { say("hello") }

    val job2 = launch { say("world") } Kotlin These jobs are resources

62. Concurrent decomposition 62 val job1 = launch { say("hello") }

    val job2 = launch { say("world") } val jobs = CompositeJob() Kotlin

63. Concurrent decomposition 63 val job1 = launch { say("hello") }

    val job2 = launch { say("world") } val jobs = CompositeJob() jobs.add(job1) jobs.add(job2) Kotlin

64. Concurrent decomposition 64 val job1 = launch { say("hello") }

    val job2 = launch { say("world") } val jobs = CompositeJob() jobs.add(job1) jobs.add(job2) jobs.cancel() Kotlin

65. Job prototype Lifetime pattern 65 interface Job : CoroutineContext.Element {

    fun cancel(reason: Throwable? = null): Boolean fun onCompletion(handler: CompletionHandler): Registration } Kotlin

66. Concurrent decomposition 66 val jobs = CompositeJob() Kotlin

67. Concurrent decomposition 67 val job = Job() Kotlin

68. Concurrent decomposition 68 val job = Job() launch(job) { say("hello")

    } launch(job) { say("world") } Kotlin Job is a coroutine context!

69. Concurrent decomposition 69 val job = Job() launch(job) { say("hello")

    } launch(job) { say("world") } job.cancel() Kotlin

70. Concurrent decomposition 70 val job = Job() launch(job) { say("hello")

    } launch(job) { say("world") } job.cancel() Kotlin !

71. Concurrent decomposition 71 val job = Job() launch(job) { say("hello")

    } launch(job) { say("world") } job.cancel() Kotlin Job becomes a CancellationToken !

72. Context propagation 72 suspend fun doSomething() { } Kotlin

73. Context propagation 73 suspend fun doSomething() { launch(coroutineContext) { say("hello")

    } launch(coroutineContext) { say("world") } } Kotlin !

74. Context propagation 74 val job = launch { launch(coroutineContext) {

    say("hello") } launch(coroutineContext) { say("world") } } Kotlin ! !

75. Context propagation 75 val job = launch(UI) { launch(coroutineContext) {

    say("hello") } launch(coroutineContext) { say("world") } } Kotlin ! ! !

76. Error propagation? 76 val job = launch { launch(coroutineContext) {

    say("hello") } launch(coroutineContext) { say("world") } } Kotlin !

77. Error propagation 77 val job = launch { say("hello") say("world")

    } Kotlin It can fail

78. Error propagation 78 val job = launch { say("hello") say("world")

    } Kotlin It can fail

79. Error propagation 79 val job = launch { launch(coroutineContext) {

    say("hello") } launch(coroutineContext) { say("world") } } Kotlin It can fail It can fail They can fail concurrently Success or failure of this composite job can be known only when all children complete cancel(ex) cancel(ex)

80. Job: the real thing 80 https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-job/index.html

81. Scope 81 val job = launch { launch(coroutineContext) { say("hello")

    } launch(coroutineContext) { say("world") } } Kotlin

82. Abstracting concurrent decomposition 82 val job = launch { launch(coroutineContext)

    { say("hello") } launch(coroutineContext) { say("world") } } Kotlin

83. Abstracting concurrent decomposition 83 val job = launch { sayHelloWorld()

    } suspend fun sayHelloWorld() { launch(coroutineContext) { say("hello") } launch(coroutineContext) { say("world") } } Kotlin

84. Abstracting concurrent decomposition 84 val job = launch { sayHelloWorld()

    } suspend fun sayHelloWorld() { launch(coroutineContext) { say("hello") } launch(coroutineContext) { say("world") } } Kotlin It can fail

85. Abstracting concurrent decomposition 85 val job = launch { sayHelloWorld()

    } suspend fun sayHelloWorld() { launch(coroutineContext) { say("hello") } launch(coroutineContext) { say("world") } } Kotlin It can fail But call returns normally! !

86. Scoping concurrency: prototype 86 val job = launch { sayHelloWorld()

    } suspend fun sayHelloWorld() { withScope { // new job in the context launch(coroutineContext) { say("hello") } launch(coroutineContext) { say("world") } } } Kotlin Throws exception on failure Encapsulated concurrent decomposition

87. Scoping concurrency: prototype 87 val job = launch { sayHelloWorld()

    } suspend fun sayHelloWorld() { withScope { launch(coroutineContext) { say("hello") } launch(coroutineContext) { say("world") } } } Kotlin

88. Scoping concurrency: prototype problems 88 val job = launch {

    sayHelloWorld() } suspend fun sayHelloWorld() { withScope { launch(coroutineContext) { say("hello") } launch(coroutineContext) { say("world") } } } Kotlin Error-pone

89. Scoping concurrency: prototype problems 89 val job = launch {

    sayHelloWorld() } suspend fun sayHelloWorld() { withScope { launch(coroutineContext) { say("hello") } launch(coroutineContext) { say("world") } } } Kotlin Error-pone Verbose

90. Scoping concurrency: solution prototype 90 val job = launch {

    sayHelloWorld() } suspend fun sayHelloWorld() { withScope { launch { say("hello") } launch { say("world") } } } Kotlin Extension function

91. Scoping concurrency: solution prototype 91 val job = launch {

    sayHelloWorld() } suspend fun sayHelloWorld() { withScope { // this: CoroutineScope launch { say("hello") } launch { say("world") } } } Kotlin Extension function

92. Scoping concurrency: solution 92 val job = launch { sayHelloWorld()

    } suspend fun sayHelloWorld() { coroutineScope { // this: CoroutineScope launch { say("hello") } launch { say("world") } } } Kotlin Extension function

93. Scoping concurrency: solution 93 val job = launch { sayHelloWorld()

    } suspend fun sayHelloWorld() { coroutineScope { // this: CoroutineScope launch { say("hello") } launch { say("world") } } } Kotlin !

94. Scoping concurrency: solution 94 suspend fun sayHelloWorld() { coroutineScope {

    // this: CoroutineScope for (w in listOf("hello", "world")) { launch { say(w) } } } }

95. Scoping concurrency: solution 95 suspend fun sayHelloWorld() { coroutineScope {

    // this: CoroutineScope for (w in listOf("hello", "world")) { launch { say(w) } } } }

96. Scoping concurrency: solution 96 suspend fun sayHelloWorld() { coroutineScope {

    // this: CoroutineScope for (w in listOf("hello", "world")) { launchSay(w) } } } fun CoroutineScope.launchSay(w: String) = launch { say(w) }

97. Scoping concurrency: solution 97 suspend fun sayHelloWorld() { coroutineScope {

    // this: CoroutineScope for (w in listOf("hello", "world")) { launchSay(w) } } } fun CoroutineScope.launchSay(w: String) = launch { say(w) }

98. Scoping concurrency: solution 98 suspend fun sayHelloWorld() { coroutineScope {

    // this: CoroutineScope for (w in listOf("hello", "world")) { launchSay(w) } } } fun CoroutineScope.launchSay(w: String) = launch { say(w) } But name for all of it?

99. 99 https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/

100. 100

101. Control flow with goto 101 https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/

102. Structured Programming 102 https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/

103. Control flow with go 103 https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/

104. Structured Concurrency 104 https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/

105. Structured Concurrency 105 coroutineScope { } launch { … }

     https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/

106. Structured Concurrency Parent always waits for children completion • Resource

     cleanup • Never loose a working coroutine • Error propagation • Never loose an exception 106

107. Structured concurrency everywhere? Similar problems, similar solutions 107

108. Structured concurrency everywhere 108

109. Structured concurrency everywhere 109 async with trio.open_nursery() as nursery: //

     … Python https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/ CoroutineScope / Job

110. Structured concurrency everywhere 110 async with trio.open_nursery() as nursery: while

     True: incoming_connection = await server_socket.accept() nursery.start_soon(connection_handler, incoming_connection) Python https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/ launch

111. Structured concurrency everywhere 111

112. Structured concurrency everywhere 112 ServerSocket listener = ... try (var

     scope = FiberScope.cancellable()) { while (...) { Socket s = listener.accept(); scope.schedule(() -> handle(s)); } } Java https://trio.discourse.group/t/project-loom-lightweight-concurrency-for-the-jvm/97 CoroutineScope / Job launch

113. Structured concurrency everywhere 113

114. Structured concurrency everywhere 114 var g errgroup.Group Go https://godoc.org/golang.org/x/sync/errgroup CoroutineScope

     / Job

115. Structured concurrency everywhere 115 var g errgroup.Group for _, url

     := range urls { // Launch a goroutine to fetch the URL. url := url // https://golang.org/doc/faq#closures_and_goroutines g.Go(func() error { resp, err := http.Get(url) if err == nil { resp.Body.Close() } return err }) } Go https://godoc.org/golang.org/x/sync/errgroup launch

116. Structured concurrency everywhere 116 var g errgroup.Group for _, url

     := range urls { // Launch a goroutine to fetch the URL. url := url // https://golang.org/doc/faq#closures_and_goroutines g.Go(func() error { resp, err := http.Get(url) if err == nil { resp.Body.Close() } return err }) } // Wait for all HTTP fetches to complete. if err := g.Wait(); err == nil { fmt.Println("Successfully fetched all URLs.") } Go https://godoc.org/golang.org/x/sync/errgroup scope completion

117. Next steps? 117 Get rid of unstructured concurrency launch {

     … }

118. Next steps? 118 Get rid of unstructured concurrency GlobalScope.launch {

     … }

119. 119 https://medium.com/@elizarov/the-reason-to-avoid-globalscope-835337445abc

120. More reading • Coroutines design document https://github.com/Kotlin/KEEP/blob/master/proposals/coroutines.md • Library guide

     https://kotlinlang.org/docs/reference/coroutines/coroutines-guide.html • Library source and issues https://github.com/Kotlin/kotlinx.coroutines • Ongoing improvement work! 120

121. Thank you Want to learn more? Questions? elizarov @ Roman

     Elizarov relizarov 121