Dynamically Parallel Sequence Operations (Map as an Example)

gistfile1.clj

(defn avg [xs] (/ (apply + xs) (count xs)))
(defn timed
  "Returns a function returning a two element vector with the execution time of f
   as the first element and the original result as the second"
  [f]
  (fn [& args]
    (let [start (. System (nanoTime))
          res (apply f args)]
      > [(/ (double (- (. System (nanoTime)) start)) 1000000.0) res])))
(defn timed-sink
  "Returns original f with sideffect of storing execution times in reference type sink"
  [f sink]
  (let [tfn (timed f)]
    (fn [& args]
      (let [[t res] (apply tfn args)]
        (swap! sink conj t)
        res))))
(def parindex {map pmap})

(defn dynamicp
  "Returns a version of f that, after tracking the first 'bound' executions and randomly choosing to parallelize,
  will return the faster (on average) of the parallel or sequential version of f. Requires f to be in parindex and
  forces f to eagerly evaluate."
  ([f] (dynamicp f 100))
  ([f bound]
   (let [stimes (atom [])
         ptimes (atom [])
         parfn (parindex f)
         tfn (timed-sink (comp doall f) stimes)
         pfn (timed-sink (comp doall parfn) ptimes)
         prn-stimes #(str "stimes: " (prn-str stimes))
         prn-ptimes #(str "ptimes: " (prn-str ptimes))]
     (fn [& args]
       (if (> (+ (count @stimes) (count @ptimes)) bound)    ; Finished sampling
         (if (> (avg @stimes) (avg @ptimes))
           (do (println "settled on parallel") (apply parfn args))
           (do (println "settled on single threaded") (apply f args)))
         (if (> 0.5 (rand))                                 ; Randomly choose tfn/pfn and track exeecution
           (do (println "sample single threaded: " (prn-stimes)) (apply tfn args))
           (do (println "sample parallel: " (prn-ptimes)) (apply pfn args))))))))

; Settles on single threaded
(let [dynamic-map (dynamicp map 10)] (dotimes [_ 20] (dynamic-map inc [1 2 3 4])))
; Settles on parallel version
(let [dynamic-map (dynamicp map 10)] (dotimes [_ 20] (dynamic-map (fn [x] (Thread/sleep 800) (inc x)) [1 2 3 4])))

(defmacro autop [[op & exprs]]
  (let [opname (gensym)]
    `(do
       (if-not (resolve (quote ~opname)) (declare ~opname))
       (if-not (bound? (var ~opname)) (def ~opname (dynamicp ~op)))
       (~opname ~@exprs))))

; Don't need to separately bind dynamic-map
(dotimes [_ 200] (autop (map inc [1 2 3])))