Select a theme

core.async

Clojure take on async programming

Created by narkisr / @narkisr

• Clojure intro
• Why async?
• core.async walkthrough
• core.async on JS

Clojure

• LISP dialect
• Concurrency tamed
• Meta programmable

Calling

 
; prefix notation
(+ 1 1) -> 1 + 1 ;

; call nesting
(println (+ 1 2 )) ->  println(1 + 2);

; interop
(.length "abc")
	    

Data structures

 
; vector
[1 2 3]

; list 
'(1 2 3)

; map 
{1 2}
	     

Defing/Scoping

 
; define a value
(def a 2)

(defn -main [name]
  (println "hello " name))

(defn scope []
  (let [a 1 b 2]
    (println a b))) ; 1 2 

	    

Control sturctures

 
(while true
  (println 1))

(if (= a 2) 
   (println "its two")
   (println "not two"))

(when foo (println "im foo"))

(loop [i 0]
 (if (= i 10)
   (println "got to 10!")
   (recur (inc i))))



	     

Why async?

• Better use of threads
• We have to (JS)
• Coordination

core.async

• Same model used in Go
• Just a library
• Targets JVM/JS

Callbacks

	    
; many to many channel
; can serve many puts/takes
(def c (chan))

; nil since its empty
(take! c (fn [v] (println v)))

; triggers the print 
(put! c "hello world")
            

But we are in callback hell!

Blocking put/take

 
(def c (chan))

;<!! means: < writes into, !! blocks and not transaction safe
(future (println "done" (>!! c 42)))

; once delivered prints 42
(future (println "Got!" (<!! c)))

; same as future but returns a channel
(println "It works!" (<!! (thread 42)))
          

But!

A thread is still required

Go macro

• A logical thread
• Turns code to a state machine
• Inspects channel operations
• Parks blocking operations
• Un-Parks when ready to run

Go

	    
; returns a many to many channel
(go 42)

(<!! (go 42))

; notice the use of <! vs <!!
(<!! (go (println "It works!" (<! (go 42)))))
         
	    

Park vs Block

• <!! >!! block a thread from a fixed thread pool
• <! >! multiplexes a logical thread across blocking calls

Buffered channels

• Back pressure mechanism
• Producer/Consumer sync

Buffered ..

 
(def fbc (chan 1)) 

(go (>! fbc 1)
    (println "done"))

(go (>! fbc 2)
    (println "done")) 

(<!! fbc)
(<!! fbc) 
	    
	    

Dropping buffers

• Don't block
• Discard messages

Dropping ..

 
; sliding-buffer drops from start
(def fbc (chan (dropping-buffer 1)))

(go (>! fbc 1)
    (println "done"))

(go (>! fbc 2)
    (println "done")) 

; we will get only 1 (2 was thrown away)
(<!! fbc)
	     

Closing

 
(def c (chan))

(close! c)
    
; we get back nil
(<!! c)
	    
	    

Real world example

Tinymasq

 
(def lookups (chan (dropping-buffer 100)))
(def answers (chan (dropping-buffer 100)))

(defn accept-loop []
  (go 
    (while true
      (let [pkt (packet (byte-array 1024))]
        (.receive @udp-server pkt)
        (>! lookups pkt)))))
	   

Processing

 
(defn process-loop []
  (go 
    (while true
      (let [pkt (<! lookups) 
            message (Message. (.getData pkt)) 
            record (.getQuestion message) 
            host (.toString (.getName record) false)
            ip (get-host (normalized-host host))]
        (when ip
          (.addRecord message (record-of host (into-bytes ip)) Section/ANSWER))
        (.setData pkt (.toWire message))
        (>! answers pkt)))))

        

Reply

 
(defn reply-loop []
  (go
    (while true
     (let [pkt (<! answers)] 
       (.send @udp-server pkt))))) 
           

Multiplexing

 
(def a (chan))
(def b (chan))

(put! a 42)                    

; will return [42 channel-with-response]
(alts!! [a b])  

(alts!! [a :default :meh]) 
	     

Timeout

 
(<!! (timeout 1000))
  
; [nil timeout-channel] 
(alts!! [a (timeout 1000)]) 

; alt with a write no takers so [nil timeout-channel] 
(alts!! [[a 42] (timeout 1000)]) 

	     

Alts ordering

 
(put! a 1)
(put! b 2)

; order will is random (prevent starving)
(alts!! [a b]) 

; we can have priority by ordering
(alts!! [a b] :priority true) 
	     

Mult and Tap

 
(def to-mult (chan 1))
(def m (mult to-mult))

(dotimes [n 4]
  (let [c (chan 1)]
    (tap m c)
    (go 
      (while true
         (when-let [v (<! c)]
           (println "Got! " v)
         (println "Exiting!"))))))

(>!! to-mult 42)
(>!! to-mult 43)

(close! to-mult)
 

Pubsub

 
(def to-pub (chan 1))
(def p (pub to-pub :type))

(let [c (chan 1)]
  (sub p :error c)
    (go 
      (while true
        (when-let [e (<! c)]
   	   (println "got an error" e)))))

(>!! to-pub {:type :error :msg "bad thing"})
(close! to-pub)
 

Clojurescript

• Clojure compiled to JS
• Not full Clojure but pretty close
• core.async supports it
• No thread just go blocks

 
(let [render (render-loop 40)]
  (loop [i 0]
    (when (< i (* width height))
      (go 
        (while true
          ; sleeping for a random time
          (<! (timeout (+ 1000 (rand-int 10000))))
          ; passing position and color 
          (>! render [(rand-int 10000) (rand-int 10)])))
      (recur (inc i)))))
       

Thank you!

@narkisr https://github.com/narkisr