Worker Queues in Clojure

In Rich Hickey’s talk “Simple Made Easy” he gives a great deal of good advice on how to manage (reduce, avoid) complexity in your code. He points out that complexity is sometimes easy to create but that simplicity is more desirable even if it is sometimes difficult. So simple != easy, and over-concentrating on developers’ ease at the expense of simplicity is killing our software.

Please, watch this talk – it’s hardly about Clojure at all, more about a rational way of thinking about coding. I’d like to concentrate on one specific piece of advice and see about how we can implement it.

Near the end, he talks about decoupling, so that if “A” wants to call a function in “B” there should be a queue of some kind in between so that A & B don’t have to be aware of each other’s existence. The coupling of A to B is unnecessary and “complects” (his word) the code.

The send

So, how to implement such a worker-queue in clojure? There is nothing much to help us in the core Clojure API but we can easily use one of Java’s BlockingQueue implementations, wrapping it up in Clojure like so:

(defn new-q [] (java.util.concurrent.LinkedBlockingDeque.))

(defn offer! 
  "adds x to the back of queue q"
  [q x] (.offer q x) q)

(defn take! 
  "takes from the front of queue q.  blocks if q is empty"
  [q] (.take q))

That’s a pretty thin wrapper! We can use it on the REPL like this:

(def example-queue (new-q))

;; background thread, blocks waiting for something to print
(future (println ">> " (take! example-queue))) 

;; the future-d thread will be unblocked and ">> HELLO WORLD" is printed.
(offer! example-queue "HELLO WORLD") 

This is fine now, we can imagine all kinds of fire&forget scenarios where we can use this like logging, writing to a database, whatever. But what if we need to have our message processed and a value returned? Can we still do this in an un-complected way?

The return

The answer is yes, and the key is in Clojure’s (promise). A promise is simply a reference-type which is initially empty and can only be written to once, using (deliver). Any attempt to deref a promise before it is delivered will block.

So, we just need to put a promise onto the queue and whatever reads the message can deliver the result. Voila:

;; THREAD 1
(def answer (promise))
(offer! example-queue {:q "how much?" :a answer})
(println @answer) ; blocks, until...

;; THREAD 2
(let [msg (take! example-queue)]
  (deliver (msg :a) "£3.50"))

;; THREAD 1
; now unblocked, prints "£3.50"

So Thread 1 fired off a message to the queue, something read the message and supplied an answer back. Thread 1 doesn’t know or care what actually did that, which simplifies it, reduces coupling and makes it more likely to be reusable. Splendid.

Concerns for real life

I’m going to try this approach in my next decent-sized project. I imagine that the entry-point to the app will def a bunch of queues and start a lot of worker threads (using future) to listen to the queues. Possibly there will be more than one listener per queue and I’ll have to implement a peek of some kind to examine waiting messages. The Java class used claims to be internally synchronised so I won’t worry about multiple deliveries etc. I may limit and monitor the size of each queue. I will use the version of deref which applies a time-limit when sending out promises, just in case.

coda

Although Mr Hickey probably isn’t interested, this approach also makes the code super-testable :)