pmapv.clj

(def pool (delay (java.util.concurrent.ForkJoinPool.)))

(defn fjtask [^Callable f]
   (java.util.concurrent.ForkJoinTask/adapt f))

(defn- fjinvoke [f]
   (if (java.util.concurrent.ForkJoinTask/inForkJoinPool)
     (f)
     (.invoke ^java.util.concurrent.ForkJoinPool
              @pool ^java.util.concurrent.ForkJoinTask
              (fjtask f))))

(defn- fjfork [task]
  (.fork ^java.util.concurrent.ForkJoinTask task))

(defn- fjjoin [task]
  (.join ^java.util.concurrent.ForkJoinTask task))

(defn pmapv [f v & [n]]
  (let [n (or n (+ 2 (.. Runtime getRuntime availableProcessors)))]
        (cond
          (empty? v) []
          (<= (count v) n) (mapv f v)
          :else
          (let [split (quot (count v) 2)
                v1 (subvec v 0 split)
                v2 (subvec v split (count v))
                fc (fn [child] #(pmapv f child n))]
            (fjinvoke
              #(let [f1 (fc v1)
                     t2 (fjtask (fc v2))]
                 (fjfork t2)
                 (into (f1) (fjjoin t2))))))))