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

core.async - Communicating Sequential Processes...

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.

Ben Mabey

April 08, 2014
Tweet

More Decks by Ben Mabey

Other Decks in Programming

Transcript

  1. 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
  2. Event loop w/Callbacks Promises/Futures/Tasks Rx (push based FRP) Coroutines &

    Generators Actors Communicating Sequential Processes (CSP) async keyword
  3. core.async is an efficient implementation of CSP that drastically reduces

    the accidental complexity usually associated with concurrent programming.
  4. core.async is an efficient implementation of CSP that drastically reduces

    the accidental complexity usually associated with concurrent programming.
  5. (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
  6. Channel Writer Writer Writer Reader Reader 1 2 3 Any

    value can be placed in channel, except for nil
  7. (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)))
  8. (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
  9. (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
  10. core.async is an efficient implementation of CSP that drastically reduces

    the accidental complexity usually associated with concurrent programming.
  11. (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)))
  12. (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)))
  13. (go (let […
 data1 (<! query1) data2 (<! query2) data3

    (<! query3)] (send-response data1 data2 data3))) Executes in pooled thread Reads value or parks
  14. (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
  15. (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)
  16. 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
  17. (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!
  18. Event loop w/Callbacks Promises/Futures/Tasks Rx (push based FRP) Coroutines &

    Generators Actors Communicating Sequential Processes (CSP) async keyword
  19. 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
  20. 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
  21. 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
  22. (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)))
  23. 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
  24. 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! !
  25. 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;! }
  26. 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.
  27. 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
  28. (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))
  29. 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.
  30. (defn start-phone-process [phone]! (let [...]! (go-loop []! (alt!! off-hook (<!

    (get-number phone))! incoming-call ...! ....! (recur))))! ! ! ! ! ! ! ! ! ! ! ! !
  31. (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*))))! ...! ))))