clj_async

clj_async.clj

(use '[clojure.algo.monads])
(require '[clojure.core.async :as async])

(defmonad async-m
          "Monad describing async operations."
          [m-result (fn m-result-async [v]
                      (let [c (async/chan)]
                        (async/go (async/>! c v))
                        c))
           m-bind   (fn m-bind-async [mv f]
                      (let [c (async/chan)]
                        (async/go
                          (let [v (async/<! mv)
                                c1 (f v)
                                final (async/<! c1)]
                            (async/>! c final)))
                        c))])
(def N 100000)

(defn calc-sync [i]
  (mod i 10))

(defn sum-sync []
  (->> (range)
       (map calc-sync)
       (take N)
       (apply +)))

(time (sum-sync))

(defn calc-async [i]
  (domonad async-m
           [- (async/timeout 0)]
           (mod i 10)))

(defn sum-async-functional []
  (->> (range)
       (map calc-async)
       (take N)
       (map async/<!!)
       (apply +)))

(time (sum-async-functional))

(defn calc-into [sum i]
  (domonad async-m
           [i (calc-async i)]
           (swap! sum #(+ i %))))

(defn sum-async-procedural []
  (let [sum (atom 0)]
    (with-monad
      async-m
      (->>
        (range)
        (map #(calc-into sum %))
        (take N)
        m-seq
        async/<!!))
    @sum))

(time (sum-async-procedural))

(defn three-things []
  (domonad
    async-m
    [a (calc-async 14)
     b (calc-async 15)
     c (calc-async 16)]
    [a b c]))

(time (async/<!! (three-things)))

(defn sum-async-procedural' []
  (let [ch (async/chan)
        sum (atom 0)]
    (async/go
      (with-monad
        async-m
        (->>
          (range)
          (map #(calc-into sum %))
          (take N)
          m-seq
          async/<!))
      (async/>! ch @sum))
    ch))

(time (async/<!! (sum-async-procedural')))