How do you block a thread until a condition becomes true?

Question

In Clojure, how do you block a thread (future) until a condition becomes true? Or, alternatively, perhaps keep retrying until a condition becomes true? This is easy when you have condition variables, but I'm not sure what's the Clojure way to do this.

To be more specific, I have a shared variable that is accessible by many futures at the same time. A future should do the following:

1. Check the state of the variable.
2. If the state meets a certain condition, update it to a new state.
3. If the state does not meet the condition, the future should block or retry, until the condition is met (by another thread modifying the state).

Answer 1

The Java platform supports condition variables, see the documentation for java.util.concurrent.locks.Condition.

The example from the above page translates easily into Clojure:

;;; based on the example in java.util.concurrent.locks.Condition
;;; documentation for JDK 1.7, see the link above

(defprotocol PBoundedBuffer
  (-put [buf x])
  (-take [buf]))

(import (java.util.concurrent.locks ReentrantLock Condition))

(deftype BoundedBuffer [^ReentrantLock lock
                        ^Condition not-full?
                        ^Condition not-empty?
                        ^objects items
                        ^:unsynchronized-mutable ^int putptr
                        ^:unsynchronized-mutable ^int takeptr
                        ^:unsynchronized-mutable ^int cnt]
  PBoundedBuffer
  (-put [buf x]
    (.lock lock)
    (try
      (while (== cnt (alength items))
        (.await not-full?))
      (aset items putptr x)
      (set! putptr (unchecked-inc-int putptr))
      (if (== putptr (alength items))
        (set! putptr (int 0)))
      (set! cnt (unchecked-inc-int cnt))
      (.signal not-empty?)
      (finally
        (.unlock lock))))

  (-take [buf]
    (.lock lock)
    (try
      (while (zero? cnt)
        (.await not-empty?))
      (let [x (aget items takeptr)]
        (set! takeptr (unchecked-inc-int takeptr))
        (if (== takeptr (alength items))
          (set! takeptr (int 0)))
        (set! cnt (unchecked-dec-int cnt))
        (.signal not-full?)
        x)
      (finally
        (.unlock lock)))))

(defn bounded-buffer [capacity]
  (let [lock (java.util.concurrent.locks.ReentrantLock.)]
    (BoundedBuffer. lock
                    (.newCondition lock)
                    (.newCondition lock)
                    (object-array capacity)
                    0
                    0
                    0)))

A test drive at the REPL:

(def bb (bounded-buffer 3))

(-put bb 1)
(-put bb 2)
(-put bb 3)

(future (-put bb 4) (println :foo))

(-take bb)

As desired, the future blocks, then prints :foo after the final call to -take.