Thread synchronization with Clojure

user_1123717 0
Viewed 610

I got an exercise:

  • Print in order all positive integers from 1 to 100.

  • Using blocks, semaphores or other similar mechanisms (but avoiding sleep), coordinate two threads such that the combined output from both threads appears in numerical order. 

     Sample Output
    In thread one: The number is ‘1’
    In thread two: The number is ‘2’
    In thread one: The number is ‘3’
    In thread two: The number is ‘4’

The exercise is for Ruby, but I'd like to show to my class that Clojure could be a good option for the task.

I don't have any experience with threads in any language, but I was thinking to use something like:

 (def thread_1 (future (swap! my-atom inc) ))
 (def thread_2 (future (swap! my-atom inc) ))

but @thread_1 always returns the same vale. Is there a way to coordinate two threads in Clojure?

I found this example in Java using ReentrantLock and Condition, and now I'm trying to translate it to Clojure.

clojure-java-interop clojure multithreading
3 Answers

If the order of threads does matter and if you are curious about non-classical thread communication, you could go with clojure.core.async and use "channels".

(require '[clojure.core.async :as a])

(let [chan-one (a/chan 1)
      chan-two (a/chan 1)]
  (a/>!! chan-one 1)
  (doseq [[thread in out] [["one" chan-one chan-two]
                           ["two" chan-two chan-one]]]
    (a/go-loop []
      (when-let [n (a/<! in)]
        (if (> n 10)
          (do (a/close! in)
              (a/close! out))
          (do (prn (format "In thread %s: The number is `%s`" thread n))
              (a/>! out (inc n))
              (recur)))))))

The output is

"In thread one: The number is `1`"
"In thread two: The number is `2`"
"In thread one: The number is `3`"
"In thread two: The number is `4`"
"In thread one: The number is `5`"
"In thread two: The number is `6`"
"In thread one: The number is `7`"
"In thread two: The number is `8`"
"In thread one: The number is `9`"
"In thread two: The number is `10`"

One gotcha here is that go-routines are executed in a thread pool, so they are not dedicated threads, and if you want to use real threads you should go this way:

(require '[clojure.core.async :as a])

(let [chan-one (a/chan 1)
      chan-two (a/chan 1)]
  (a/>!! chan-one 1)
  (doseq [[thread in out] [["one" chan-one chan-two]
                           ["two" chan-two chan-one]]]
    (a/thread
      (loop []
        (when-let [n (a/<!! in)]
          (if (> n 10)
            (do (a/close! in)
                (a/close! out))
            (do (prn (format "In thread %s: The number is `%s`" thread n))
                (a/>!! out (inc n))
                (recur))))))))
user_2986240

First, the reason you're seeing odd results is you're dereferencing the future, not the atom holding the number. The future will return the result of swap! when dereferenced.

Second, you can use locking (basically Java's synchronized) to only allow one thread to increase and print at a time:

(def n-atom (atom 0))

(defn action []
  ; Arbitrarily chose the atom to lock on.
  ; It would probably be better to create a private object that can't be otherwise used.
  (locking n-atom
    (println
      (swap! n-atom inc))))

(defn go []
  ; Have each thread do action twice for a total of four times
  (future (doall (repeatedly 2 action)))
  (future (doall (repeatedly 2 action))))

(go)
1
2
3
4

I'll note though that future really shouldn't be used here. future is for when you want to compute a result asynchronously. It swallows errors until it's dereferenced, so if you never @, you'll never see exceptions that come up inside the future. It would be better to use a thread pool, or, with the help of a macro for ease-of-use, start up two threads yourself:

(defmacro thread
  "Starts the body in a new thread."
  [& body]
  `(doto (Thread. ^Runnable (fn [] ~@body))
         (.start)))

(defn go []
  (thread (doall (repeatedly 2 action)))
  (thread (doall (repeatedly 2 action))))
user_30002060

Here's a way to generally coordinate multiple threads operating on the same state using an agent:

(def my-agent (agent 1 :validator #(<= % 100)))
(future
  (while true
    (send my-agent
          (fn [i]
            (println "In thread" (.getName (Thread/currentThread))
                     "the number is:" i)
            (inc i)))))
In thread clojure-agent-send-pool-4 the number is: 1
In thread clojure-agent-send-pool-5 the number is: 2
In thread clojure-agent-send-pool-5 the number is: 3
In thread clojure-agent-send-pool-4 the number is: 4
In thread clojure-agent-send-pool-4 the number is: 5
In thread clojure-agent-send-pool-4 the number is: 6

You'd see the same essential behavior here with or without the future because send returns immediately in the inner loop, and the sent function may be executed on different threads in the pool. The agent takes care of coordinating access to the shared state.

Update: here's another way to do the same thing that doesn't involve a :validator function, or termination by exception:

(def my-agent (agent (range 1 101)))
(while (seq @my-agent)
  (send
    my-agent
    (fn [[n & ns]]
      (when n
        (println "In thread" (.getName (Thread/currentThread))
                 "the number is:" n)
        ns))))
user_87462160
Related