A small comparison of many to many thread-safe queues in Clojure

many-to-many-queues.clj

(import 'java.util.concurrent.LinkedBlockingQueue)
(use 'criterium.core)

(defn lbq-test []
  (let [q ^LinkedBlockingQueue (LinkedBlockingQueue.)]
    (.start(Thread. #(dotimes [n 50000] (.take q))))
    (.start(Thread. #(dotimes [n 50000] (.take q))))
    (.start(Thread. #(dotimes [n 50000] (.take q))))
    
    (.start(Thread. #(dotimes [n 1000000] (.add q n))))
    (.start(Thread. #(dotimes [n 1000000] (.add q n))))

    (dotimes  [n 50000] (.take q))))


(defn try-deliver [a v]
  (try
    (deliver (first a) v)
    (catch Exception e)))

(defn dropbuf-test []
  (let [a (atom [])]

    (.start(Thread. #(dotimes [n 50000] 
      (let [p (promise)]
        (swap! a conj p) @p))))
    (.start(Thread. #(dotimes [n 50000]  
      (let [p (promise)]
        (swap! a conj p) @p))))
    (.start(Thread. #(dotimes [n 50000]  
      (let [p (promise)]
        (swap! a conj p) @p))))


    (.start(Thread. 
            #(dotimes [n 1000000]
               (swap! a (fn [x] (try-deliver x n) (rest x))))))
    (.start(Thread. 
            #(dotimes [n 1000000]
               (swap! a (fn [x] (try-deliver x n) (rest x))))))

    (dotimes [n 50000]  
      (let [p (promise)]
        (swap! a conj p)@p))))


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

(defn ca-test []
  (let [c (async/chan 2000000)]
    (async/thread(dotimes [n 50000] (async/<!! c)))
    (async/thread(dotimes [n 50000] (async/<!! c)))
    (async/thread(dotimes [n 50000] (async/<!! c)))
    
    (async/thread(dotimes [n 1000000] (async/>!! c n)))
    (async/thread(dotimes [n 1000000] (async/>!! c n)))

    (dotimes [n 50000] (async/<!! c))))

;fill rate
(defn fill-atom []
  (let [a (atom [])]
    (.start(Thread. #(dotimes [n 1000000] (swap! a conj n))))
    (dotimes [n 1000000] (swap! a conj n))))


(defn fill-lbq [] 
  (let [q ^LinkedBlockingQueue (LinkedBlockingQueue.)]
    (.start(Thread. #(dotimes [n 1000000] (.add q n))))
    (dotimes [n 1000000] (.add q n))))