clojure-dev

Conclusion first: It turns out that Clojure's concurrency primitives allow you, with a very moderate amount of uncomplicated code, to take advantage of parallel hardware and outperform really fast software when it doesn't take such advantage.

These, "refs" for short, are one of the three tools offered by Clojure to make concurrency practical and manageable. Herewith a walk-through of code that uses them to accomplish a simple task in a highly concurrent fashion.

;; Here's an example distilled from real-world code:

(defn p [string]
(for [mime-exts (.split string ",")
:let [[mime & exts] (.split mime-exts ";")]]
[(keyword mime) exts]))


;; With -> destructuring, could be written like this:

(defn p [string]
(for [(-> (.split ";") [mime & exts]) (.split string ",")]
[(keyword mime) exts]))


;; Either way, it does this:

user=> (p "text/plain;foo;bar,text/xml;extra")
([:text/plain ("foo" "bar")] [:text/xml ("extra")])

(defn pipe
"Returns a pair: a seq (the read end) and a function (the write end).
The function can takes either no arguments to close the pipe
or one argument which is appended to the seq. Read is blocking."
[]
(let [promises (atom (repeatedly promise))
p (second @promises)]
[(lazy-seq @p)
(fn
([] ;close the pipe
(let [[a] (swap! promises #(vector (second %)))]
(if a
(deliver a nil)
(throw (Exception. "Pipe already closed")))))
([x] ;enqueue x
(let [[a b] (swap! promises next)]
(if (and a b)
(do
(deliver a (cons x (lazy-seq @b)))
x)
(throw (Exception. "Pipe already closed"))))))]))

Beware of not printing the seq while the pipe is still open!

(let [[q f] (pipe)]
(future (doseq [x q] (println x)) (println "that's all folks!"))
(doseq [x (range 10)]
(f x))
(f)) ;close the pipe

Clojars.org will be a dead easy jar repository for open source Clojure libraries.

clojure.core/extend-protocol
([p & specs])
Macro
Useful when you want to provide several implementations of the same
protocol all at once. Takes a single protocol and the implementation
of that protocol for one or more types. Expands into calls to
extend-type and extend-class:

(extend-protocol Protocol
::AType
(foo [x] ...)
(bar [x y] ...)
::BType
(foo [x] ...)
(bar [x y] ...)
AClass
(foo [x] ...)
(bar [x y] ...)
nil
(foo [x] ...)
(bar [x y] ...))

expands into:

(do
(clojure.core/extend-type ::AType Protocol
(foo [x] ...)
(bar [x y] ...))
(clojure.core/extend-type ::BType Protocol
(foo [x] ...)
(bar [x y] ...))
(clojure.core/extend-class AClass Protocol
(foo [x] ...)
(bar [x y] ...))
(clojure.core/extend-type nil Protocol
(foo [x] ...)
(bar [x y] ...)))

(defn hydra
"returns a BlockingQueue, will return a new infinite lazy-seq wrapped in a delay
evertime it is deref'ed. all items put in the LinkedBlockingQueue will be added to
all the lazy-seqs producded by deref'ing"
[]
(let [consumers (atom nil)
producer (proxy [LinkedBlockingQueue IDeref] []
(deref []
(let [x (LinkedBlockingQueue.)]
(swap! consumers conj x)
(delay (repeatedly #(.take x))))))]
(future
(while true
(let [x (.take producer)]
(doseq [y @consumers]
(.put y x)))))
producer))