core.async - Communicating Sequential Processes in Clojure

Talk given at Lambda Lounge on April 8, 2014 http://www.meetup.com/Lambda-Lounge-Utah/events/164368202/

Every language these days seems to be adding some kind of asynchronous support. With all of the available options and buzzwords it is easy to lose sight of what problems we are trying to solve and how these new approaches address all the pain points. We'll begin by motivating the need for asynchronous programming and discuss the approaches typically taken (e.g. callbacks, promises) along with associated limitations and trade-offs. Communicating Sequential Processes (CSP) will then be introduced via Clojure's core.async library and show how this model help reduces the complexity of working with asynchronous events. We will also look at the implementation of core.async, how it adds the functionality as a library via macros and what trade-offs come along with that decision.

Transcript

1. Ben Mabey Communicating Sequential Processes in Clojure core.async @bmabey Lambda Lounge Utah, April 8 2014
2. Everything is a Remix … but especially this presentation.
3. Everything is a Remix … but especially this presentation. https://speakerdeck.com/hlship/mastering-time-with-clojure-core-dot-async Mastering Time with Clojure core.async by Howard M. Lewis Ship http://www.infoq.com/presentations/Death-by-Accidental-Complexity Death by Accidental Complexity by Ulf Wiger https://github.com/stuarthalloway/presentations/wiki/Core-Async core.async by Stuart Halloway
4. Event loop w/Callbacks Promises/Futures/Tasks Rx (push based FRP) Coroutines & Generators Actors Communicating Sequential Processes (CSP) async keyword
5. core.async is an efficient implementation of CSP that drastically reduces the accidental complexity usually associated with concurrent programming.
6. core.async is an efficient implementation of CSP that drastically reduces the accidental complexity usually associated with concurrent programming.
7. Embarrassingly Parallel …emparallel? aka brain dead concurrency
8. (defn request-handler! [request]! (let [data1 (do-query-1 request)! data2 (do-query-2 request)! data3 (do-query-3 request)]! (send-response data1 data2 data3))) Embarrassingly Parallel …emparallel? aka brain dead concurrency
9. 1 3 2
10. 1 3 2
11. 1 3 2
12. Channel Writer Writer Writer Reader Reader
13. Channel Writer Writer Writer Reader Reader 1 2 3
14. Channel Writer Writer Writer Reader Reader 1 2 3 Any value can be placed in channel, except for nil
15. Channel Writer Writer Writer Reader Reader 1 2 3
16. Channel Writer Writer Writer Reader Reader 1 2 3
17. Channel Writer Writer Writer Reader Reader 1 2 3
18. Channel Writer Writer Writer Reader Reader 3
19. Channel Writer Writer Writer Reader Reader
20. Channel Reader Reader (close! channel)
21. Channel Reader Reader (close! channel) nil nil
22. Channel Writer Writer Writer
23. Channel Writer Writer Writer 1 2 3 4
24. Channel Writer Reader Reader Writer Writer 1 2 3 4
25. Channel Writer Reader Reader Writer Writer 1 2 3 4
27. (defn request-handler! [request]! (let [data1 (do-query-1 request)! data2 (do-query-2 request)! data3 (do-query-3 request)]! (send-response data1 data2 data3)))
28. (defn request-handler [request] (let [query1 (thread (do-query-1 request)) query2 (thread (do-query-2 request)) query3 (thread (do-query-3 request)) data1 (<!! query1) data2 (<!! query2) data3 (<!! query3)] (send-response data1 data2 data3))) Evaluate in other thread, return channel
29. (defn request-handler [request] (let [query1 (thread (do-query-1 request)) query2 (thread (do-query-2 request)) query3 (thread (do-query-3 request)) data1 (<!! query1) data2 (<!! query2) data3 (<!! query3)] (send-response data1 data2 data3))) block thread until value available
30. (thread (do-query-1 request)) (<!! query1) Result – (chan 1) (do-query-1 request)
31. 1 3 2
32. core.async is an efficient implementation of CSP that drastically reduces the accidental complexity usually associated with concurrent programming.
35. (defn request-handler [request] (let [query1 (thread (do-query-1 request)) query2 (thread (do-query-2 request)) query3 (thread (do-query-3 request)) data1 (<!! query1) data2 (<!! query2) data3 (<!! query3)] (send-response data1 data2 data3)))
36. (defn request-handler [request] (let [result (go (let [query1 (thread (do-query-1 request)) query2 (thread (do-query-2 request)) query3 (thread (do-query-3 request)) data1 (<! query1) data2 (<! query2) data3 (<! query3)] (send-response data1 data2 data3)))] (<!! result)))
37. (go (let […
 data1 (<! query1) data2 (<! query2) data3 (<! query3)] (send-response data1 data2 data3))) Executes in pooled thread Reads value or parks
38. (go (let […
 data1 (<! query1) data2 (<! query2) data3 (<! query3)] (send-response data1 data2 data3))) Executes in pooled thread Reads value or parks (defn request-handler [request] (let [result (go …)] (<!! result))) Final result available through channel
39. 1 3 2
40. 1 3 2
41. (defn request-handler [request] (go (let [query1 (thread (do-query-1 request)) query2 (thread (do-query-2 request)) query3 (thread (do-query-3 request)) data1 (<! query1) data2 (<! query2) data3 (<! query3)] (send-async-response request data1 data2 data3))) nil)
42. query1 (thread (do-query-1 request)) query2 (thread (do-query-2 request)) query3 (thread (do-query-3 request)) 0 data1 (<! query1) 1 data2 (<! query2) 2 data3 (<! query3)
 
 (send-async-response request data1 data2 data3) 3
45. (defn google [query]! (let [results-chan (chan)! timeout-chan (timeout 100)]! (go (>! results-chan (<! (web query))))! (go (>! results-chan (<! (image query))))! (go (>! results-chan (<! (video query))))! ! ! ! ! (go-loop [i 0 results []]! (if (= i 3)! results! (recur (inc i)! (->>! (alt! [results-chan timeout-chan] ([v] v))! (conj results))))))) Shared timeout!
46. Event loop w/Callbacks Promises/Futures/Tasks Rx (push based FRP) Coroutines & Generators Actors Communicating Sequential Processes (CSP) async keyword
48. var  tweets,  answers,  kaggleProfile;
  
 twitter.getTweetsFor("bmabey",  function  (result)  {
    tweets  =  result;
    somethingFinished();
 });
 stackOverflow.getAnswersFor("bmabey",  function  (result)   {
    answers  =  result;
    somethingFinished();
 });
 kaggle.getProfileOf("bmabey",  function  (result)  {
    kaggleProfile  =  result;
    somethingFinished();
 }); var  finishedSoFar  =  0;
  
 function  somethingFinished()  {
    if  (++finishedSoFar  ===  3)  {
        ui.show(tweets,  answers,  kaggleProfile);
    }
 } http://www.slideshare.net/domenicdenicola/callbacks-promises-and-coroutines-oh-my-the-evolution-of-asynchronicity-in-javascript
49. var  tweets,  answers,  kaggleProfile;
  
 twitter.getTweetsFor("bmabey",  function  (result)  {
    tweets  =  result;
    somethingFinished();
 });
 stackOverflow.getAnswersFor("bmabey",  function  (result)   {
    answers  =  result;
    somethingFinished();
 });
 kaggle.getProfileOf("bmabey",  function  (result)  {
    kaggleProfile  =  result;
    somethingFinished();
 }); var  finishedSoFar  =  0;
  
 function  somethingFinished()  {
    if  (++finishedSoFar  ===  3)  {
        ui.show(tweets,  answers,  kaggleProfile);
    }
 } http://www.slideshare.net/domenicdenicola/callbacks-promises-and-coroutines-oh-my-the-evolution-of-asynchronicity-in-javascript
50. var  tweets,  answers,  kaggleProfile;
  
 twitter.getTweetsFor("bmabey",  function  (result)  {
    tweets  =  result;
    somethingFinished();
 });
 stackOverflow.getAnswersFor("bmabey",  function  (result)   {
    answers  =  result;
    somethingFinished();
 });
 kaggle.getProfileOf("bmabey",  function  (result)  {
    kaggleProfile  =  result;
    somethingFinished();
 }); var  finishedSoFar  =  0;
  
 function  somethingFinished()  {
    if  (++finishedSoFar  ===  3)  {
        ui.show(tweets,  answers,  kaggleProfile);
    }
 } - Continuation Passing Style - Destroys readability - Forces shared mutable state - Can’t use normal language constructs for error handling and looping. http://www.slideshare.net/domenicdenicola/callbacks-promises-and-coroutines-oh-my-the-evolution-of-asynchronicity-in-javascript
51. (go! (let [results (chan 3)]! (twitter/get-tweets "bmabey" c)! (stack-overflow/get-answers "bmabey" c)! (kaggle/get-profile "bmabey" c)! (send-async-response (async/into [] results)))
53. Q.all([
    twitter.getTweetsFor("bmabey"),
    stackOverflow.getAnswersFor("bmabey"),
    kaggle.getProfileOf("bmabey")   ]).then(function  (results)  {
    console.log(results[0],  results[1],  results[2]);
 }); http://www.slideshare.net/domenicdenicola/callbacks-promises-and-coroutines-oh-my-the-evolution-of-asynchronicity-in-javascript
54. Q.all([
    twitter.getTweetsFor("bmabey"),
    stackOverflow.getAnswersFor("bmabey"),
    kaggle.getProfileOf("bmabey")   ]).then(function  (results)  {
    console.log(results[0],  results[1],  results[2]);
 }); http://www.slideshare.net/domenicdenicola/callbacks-promises-and-coroutines-oh-my-the-evolution-of-asynchronicity-in-javascript + Promises are values + Composable + Better error handling - Continuation Passing Style! - Still can’t use language constructs. - Mutable state still flying around. - By design, are not for streams! - Have to create a new promise for each event.. every time! !
55. C# Async async void Go() {! _button.IsEnabled = false;! string[] urls = "clojure.org www.albahari.com/nutshell/ golang.org".Split();! int totalLength = 0;! foreach (string url in urls)! {! var uri = new Uri ("http://" + url);! byte[] data = await new WebClient().DownloadDataTaskAsync (uri); _results.Text += "Length of " + url + " is " + data.Length +! totalLength += data.Length;! }! _results.Text += "Total length: " + totalLength;! }
56. C# Async async void Go() {! _button.IsEnabled = false;! string[] urls = "clojure.org www.albahari.com/nutshell/ golang.org".Split();! int totalLength = 0;! foreach (string url in urls)! {! var uri = new Uri ("http://" + url);! byte[] data = await new WebClient().DownloadDataTaskAsync (uri); _results.Text += "Length of " + url + " is " + data.Length +! totalLength += data.Length;! }! _results.Text += "Total length: " + totalLength;! } + No more Continuation Passing Style! - Still can’t use for streams.
57. Async Adoption 2007 - F# adds introduces async workflows 2010? - C# adds async 2012-13? - Scala library adds async via macros 2013 - Python announced it will be adding it 2014 - Javascript, ES7 proposal for async
59. (defn batch-until-pause! "Batches values from the `in` channel into a single a vec of values up until no! messages have appeared in `timeout-pause` milliseconds. `max-batch-time` is the! number of milliseconds to wait overall before sending a batch."! ([timeout-pause max-batch-time in]! (batch-until-pause timeout-pause max-batch-time in (chan)))! ([timeout-pause max-batch-time in out]! (go! (loop [batch []! batch-timeout (timeout max-batch-time)]! (alt!! ;; batch timed out, send what we have now and start a new batch timeout! batch-timeout (do! (>! out batch)! (recur [] (timeout max-batch-time)))! in ([v]! (if v! ;; we got a value, add it to the batch! (recur (conj batch v) batch-timeout)! ;; we got nil so the 'in' channel was closed, send our batch! (>! out batch)))! ! ;; a timeout happened, send the batch and start a new one! (timeout timeout-pause) (do! (>! out batch)! (send-batch!)! (recur [] batch-timeout))))! (close! out))! out))
60. What about Rx?
61. What about Rx? + Very composable + Handles streams - Push-based semantics only cover ~80% use cases - Have to learn all the combinators - Writing a combinator can be difficult. - Mutable state with call backs again! ! Awesome abstraction, but in the end it is more complex and less powerful than CSP.
64. We’ve covered brain dead concurrency… what about mind blowing concurrency?
67. (defn start-phone-process [phone]! (let [...]! (go-loop []! (alt!! off-hook (<! (get-number phone))! incoming-call ...! ....! (recur))))! ! ! ! ! ! ! ! ! ! ! ! !
68. (defn start-phone-process [phone]! (let [...]! (go-loop []! (alt!! off-hook (<! (get-number phone))! incoming-call ...! ....! (recur))))! ! ! (defn get-number [phone]! (let [...]! (go-loop [numbers []]! (alt!! on-hook ::hang-up! digit-pressed ([num] (let [numbers* (conj numbers num)]! (if (= numbers* 9)! (<! (dial-number numbers* phone))! (recur numbers*))))! ...! ))))
74. Differences with Actors…
75. Macro Limitations
76. Thank you! ! ! BenMabey.com ! github.com/bmabey ! @bmabey !