Java ForkJoinPool in Clojure, plus benchmark

forkjoin.clj

(ns forkjoin
  (:import [java.util.concurrent ForkJoinPool ForkJoinTask RecursiveTask]))

(defonce ^ForkJoinPool pool (ForkJoinPool/commonPool))

;; Simplest possible wrappers for ForkJoin API

(defn task [f]
  (proxy [RecursiveTask] []
    (compute []
      (f))))

(defn fork [^ForkJoinTask task]
  (if (ForkJoinTask/inForkJoinPool)
    (.fork task)
    (.execute pool task))
  task)

;; Example: three tasks in parallel

(defn sleep-and-return [num]
  (Thread/sleep num)
  num)

(defn fork-join-example []
  (let [a (fork (task #(sleep-and-return 500)))
        b (fork (task #(sleep-and-return 600)))
        c (fork (task #(sleep-and-return 700)))]
    {:result-a @a ;; deref is .join()
     :result-b @b
     :result-c @c}))

(comment
  (time (fork-join-example))
  ;; "Elapsed time: 700.692888 msecs"
  ;; {:result-a 500, :result-b 600, :result-c 700}
  )

;; Example: data driven concurrency: evaluating a tree

(defn prn-atomic [& args]
  (locking *out*
    (apply prn args)))

(defn eval-one [node]
  (assoc node :value (conj (mapv :value (:children node)) (:value node))))

(defn eval-all-impl [node]
  (let [child-tasks (mapv eval-all-impl (:children node))]
    (prn-atomic :QUEUED (:value node))
    (fork
     (task
      (fn []
        (prn-atomic :STARTED (:value node))
        (let [result (-> node
                         (assoc :children (mapv deref child-tasks))
                         eval-one)]
          (prn-atomic :DONE (:value node))
          result))))))

(defn eval-all [node]
  @(eval-all-impl node))

(comment
  (:value
   (eval-all {:sleep 0
              :value :parent
              :children [{:sleep 100
                          :value :child
                          :children [{:sleep 500
                                      :value :a}
                                     {:sleep 100
                                      :value :b}]}
                         {:sleep 0
                          :value :child2
                          :children [{:sleep 200
                                      :value :c}
                                     {:sleep 300
                                      :value :d}]}]}))
  ;; => [[[:a] [:b] :child] [[:c] [:d] :child2] :parent]
  )

;; Example: benchmarking against futures

(defn future-sum [v]
  (let [n (count v)]
    (if (< n 5)
      (apply + v)
      (let [half (int (/ n 2))
            left (subvec v 0 half)
            right (subvec v half)
            left-sum (future (future-sum left))
            right-sum (future (future-sum right))]
        (+ @left-sum @right-sum)))))

(defn forkjoin-sum [v]
  (let [n (count v)]
    (if (< n 5)
      (apply + v)
      (let [half (int (/ n 2))
            left (subvec v 0 half)
            right (subvec v half)
            left-sum (fork (task #(forkjoin-sum left)))
            right-sum (fork (task #(forkjoin-sum right)))]
        (+ @left-sum @right-sum)))))

(comment
  (def data (vec (range 100000)))
  (time (future-sum data))
  ;; around 60ms on my machine
  (time (forkjoin-sum data))
  ;; around 2ms on my machine
  )