Parallel processing

parallel_processing.clj

;; Experiment - parallel processing using missionary primitives.
;; Inspired by Rx's parallel 'rails' :
;; http://reactivex.io/RxJava/3.x/javadoc/io/reactivex/rxjava3/parallel/ParallelFlowable.html
;; https://dzone.com/articles/rxjava-idiomatic-concurrency-flatmap-vs-parallel

(ns parallel-processing
  (:require [missionary.core :as m]))

(defn map-task [f >x]
  (m/ap (m/? (f (m/?> >x)))))

(defmacro do-after [& body]
  `(fn [rf#]
     (fn ([r#] ~@body (rf# r#))
       ([r# x#] (rf# r# x#)))))

(defn poll [t]
  (m/ap (m/? (m/?> (m/seed (repeat t))))))

(defn parallel
  "Process values of flow >x with n parallel instances of pipeline f."
  [n f >x]
  (assert (pos? n))
  (m/ap
    (let [r (m/rdv) d (m/dfv) c (m/race r d)]
      (m/?> (m/?= (m/seed (cons (->> >x
                                  (map-task r)
                                  (m/eduction (filter {}) (do-after (d c))))
                            (repeat n
                              (->> (poll c)
                                (m/eduction (take-while (complement #{c})))
                                (f))))))))))

;; returns 499500 in 2 seconds
(m/? (->> (m/seed (range 1000))
       (parallel 50 #(m/ap (m/? (m/sleep 100 (m/?> %)))))
       (m/reduce +)))