core.async thoughts

blocking.clj

;; ------------------------

(defn- log-time [{:keys [ns name line]} f & args]
  (let [start   (System/nanoTime)
        res     (apply f args)
        elapsed (quot (- (System/nanoTime) start) 1000)]
    (log/debug (format "%s/%s:%s %dus" ns name line elapsed))
    res))

(defn enable-timing [var]
  (log/debug "enabling timings" var)
  (robert.hooke/add-hook var (partial log-time (meta var))))

(defn enable-timing-on-blocking-functions []
  (->> (all-ns)
       (mapcat ns-interns)
       (map second)
       (filter #(:blocking (meta %)))
       ;; No dups
       (remove #(:robert.hooke/hook (meta (deref %))))
       (map enable-timing)
       doall))

;; ------------------------

(defn- ^:blocking request-data [s]
  (client/get (str "http://some.query?q=s") {:throw-exceptions true}))

(defn get-data [s]
  (async/go (try 
              (request-data s)
              ;; catch and put exception on the channel
              (catch Exception e
                w))))

get-result-go-better.clj

(defn get-result-go-better []
  (->>
   (repeatedly 1000 calculation-go)
   (async/merge)
   (async/reduce + 0)))

(time (async/<!! (get-result-go-better)))
;; "Elapsed time: 178.654 msecs"

get-result-go-macro.clj

(defmacro calculation-go-macro []
  `(do
    ;; wait for some stuff using async/channels
     (rand-int 1000)))

(defn get-result-go-macro []
  (async/go
    ;; wait for some stuff
    (reduce +
            (loop [res [] cnt 1000]
              (if (zero? cnt)
                res
                (recur (conj res (calculation-go-macro)) (dec cnt)))))))

(time (async/<!! (get-result-go-macro)))
;; "Elapsed time: 2.452 msecs"

get-result-go.clj

(defn calculation-go []
  (async/go
    ;; wait for some stuff
    (rand-int 1000)))

(defn get-result-go []
  (async/go
    (->>
     (repeatedly 10 calculation-go)
     (map async/<!)
     (reduce +))))

(async/<!! (get-result-go))
;; => nil

get-result-go2.clj

(defn calculation-go2 [ch]
  (async/go
    ;; wait for some stuff
    (>! ch (rand-int 1000))))

(defn get-result-go2 []
  (let [ch (async/chan)
        count 1000]
    (dotimes [_ count]
        (calculation-go2 ch))
    (async/go-loop [res 0 cnt count]
      (if (or (nil? res) (zero? cnt))
        res
        (recur (+ res (<! ch)) (dec cnt))))))

(time (async/<!! (get-result-go2)))
;; "Elapsed time: 102.503 msecs"

get-result.clj

(defn calculation []
  ;; Big complicated calculation
  (rand-int 1000))

(defn get-result []
  (async/go
    ;; wait for some stuff
    (reduce + (repeatedly 1000 calculation))))

(time (async/<!! (get-result)))
;; "Elapsed time: 1.034 msecs"

getResult.go

func calculation() int {
	return rand.Intn(1000)
}

func getResult(c chan int) {
	sum := 0
	for i := 0; i < 1000; i++ {
		sum += calculation()
	}
	c <- sum
}
// Elapsed time: 196.309us

func calculation2(c chan int) {
	// wait for some stuff
	c <- rand.Intn(1000)
}

func getResult2(c chan int) {
	tmpc := make(chan int)
	sum := 0
	for i := 0; i < 1000; i++ {
		go calculation2(tmpc)
	}
	for i := 0; i < 1000; i++ {
		sum += <-tmpc
	}
	c <- sum
}
// Elapsed time: 4.421806ms

throw-err.clj

(defn throw-err [e]
  (when (instance? Throwable e) (throw e))
  e)

(defmacro <? [ch]
  `(throw-err (async/<! ~ch)))

(defn get-data [s]
  (async/go (try
              (client/get (str "http://some.query?q=s") {:throw-exceptions true})
              ;; catch and put exception on the channel
              (catch Exception e
                e))))

(async/go (try
            (let [data      (<? (get-data "clojure"))
                  more-data (<? (get-data "core.async")]
              ;; process all data
              )
            ;; Handle exceptions for all '<? calls'
            (catch Exception e
              (log/error "error getting data"))))