Performance of ConcurrentLinkedQueue + Semaphore vs atom + persistent_vector for queue implementation

kafka-clj.pool-impl.clj

(ns
  ^{:doc "Concurrent keyed pool implementation using ConcurrentHashMap and ConcurrentLinkedQueue"}
  kafka-clj.pool-impl
  (:use criterium.core)
  (:import (java.util.concurrent ConcurrentLinkedQueue ConcurrentHashMap Semaphore ExecutorService Executors TimeUnit)))

;; add [criterium "0.4.4"] to you're project.clj file
;; then use  run-test-cc and run-test-a
;;
;; Results for both ConcurrentLinkedQueue + Semaphore and for atom + vector access is the same
;; which brings me the the conclusion that its ok in terms of performance to use atomis and vectors for this use case.


(defn ^ConcurrentLinkedQueue queue []
  (ConcurrentLinkedQueue.))

(defn ^ConcurrentHashMap ccmap []
  (ConcurrentHashMap.))

(defn cc-pool []
  {:queue (queue) :sem (Semaphore. 100)})

(defn cc-poll [{:keys [^ConcurrentLinkedQueue queue ^Semaphore sem]}]
  (.acquire sem)
  (.poll queue))

(defn cc-return [{:keys [^ConcurrentLinkedQueue queue ^Semaphore sem]} v]
  (.offer queue v)
  (.release sem))

(defn a-pool []
  {:queue (atom [nil []])})

(defn a-poll [{:keys [queue]}]
  (nth (swap! queue (fn [[_ [x & xs]]] [x xs])) 0))

(defn a-return [{:keys [queue]} v]
  (swap! queue (fn [[x xs]] [x (vec (conj xs v))])))

(defn test-threads [n i f]
  (let [^ExecutorService service (Executors/newFixedThreadPool n)
        f2 (fn []
             (try
               (dotimes [_ i]
                 f)
               (catch Exception e (.printStackTrace e))))]

    (dotimes [_ n]
      (.submit service ^Runnable f2))

    (.shutdown service)
    (.awaitTermination service 10 TimeUnit/MINUTES)))


(defn test-cc []
  (let [pool (cc-pool)]
    (test-threads 8 10000 (fn []
                            (cc-return pool (cc-poll pool))))))

(defn test-a []
  (let [pool (a-pool)]
    (test-threads 8 10000 (fn []
                            (a-return pool (a-poll pool))))))


(defn run-test-cc []
  (with-progress-reporting (bench (test-cc) :verbose)))

(defn run-test-a []
  (with-progress-reporting (bench (test-a) :verbose)))