Which is the ideal Clojure concurrency construct for maintaining queue of data to be processed?

Question

If I want to maintain a queue of frames of images on the server side that I will be sending to the client what data structure should I use ?

I am trying to make a simple app where I will send frames to the server and server will then push them to other clients.

should I maintain this queue as an atom or as a ref?

Answer 1

You could just use one of the queue classes from java.util.concurrent. Easy access to the Java standard library is after all one of the strong points of Clojure, so you don't have to build everything yourself from the building blocks provided by Clojure if Java already provides something that does the job.

I would suggest to choose something from the implementations of the interface BlockingQueue.

Answer 2

The most important operation of a queue is popping items: namely getting-and-removing an item from the collection.

As this operation is composite, refs are more naturally suited to perform it atomically, this is - preventing race conditions (e.g. two threads getting the same item).

(defn remove-and-get [queue]
  (dosync
   (let [i (peek @queue)]
     (alter queue pop)
     i)))

(def q (ref (clojure.lang.PersistentQueue/EMPTY)))

(dosync
 (commute q conj 42)
 (commute q conj :foo)
 (commute q conj []))

[(seq @q) (remove-and-get q) (seq @q)]
;; evals to [(42 :foo []) 42 (:foo [])]

Equivalent functionality can be implemented in terms of atoms as well.

(defn remove-and-get [queue]
  (let [snapshot @queue
        i (peek snapshot)]
    (if (compare-and-set! queue snapshot (pop snapshot))
      i
      (recur queue))))

(def q (atom (clojure.lang.PersistentQueue/EMPTY)))

(swap! q conj 42)
(swap! q conj :foo)
(swap! q conj [])

[(seq @q) (remove-and-get q) (seq @q)]

Answer 3

Seems to be a possibility for core.async.