Structured Concurrency

A traditional approach to concurrency in programming languages is well known — you are given primitives that launch concurrent processes, threads, tasks, coroutines, actors (you name it!) and some means to establish communication between them. As systems become more concurrent and as those primitives become more lightweight, tracking lifetimes of those entities and making sure they do not leak becomes a challenge.

In this talk we'll tell the story of the path that we went through when designing concurrency libraries for Kotlin programming language. How we started with traditional concurrency primitives and how we've discovered and implemented the concept of structured concurrency, what's behind the name, and how it is now gaining popularity in other ecosystems. It would not be an exaggeration to say that we are witnessing a programming style revolution akin to move from GOTO-based unstructured code of the past to the structured programming paradigms of today that had started with iconic "Go To Statement Considered Harmful" letter of Dijkstra in 1968.

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/
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