;   Copyright (c) Rich Hickey. All rights reserved.

 ;   The use and distribution terms for this software are covered by the

 ;   Eclipse Public License 1.0 (http://opensource.org/licenses/eclipse-1.0.php)

 ;   which can be found in the file epl-v10.html at the root of this distribution.

 ;   By using this software in any fashion, you are agreeing to be bound by

 ;   the terms of this license.

 ;   You must not remove this notice, or any other, from this software.

 

 (ns ^{:doc "DEPRECATED Wrapper of the ForkJoin library (JSR-166)."

        :author "Rich Hickey"}

     clojure.parallel)

 (alias 'parallel 'clojure.parallel)

 

 (comment "

 The parallel library wraps the ForkJoin library scheduled for inclusion in JDK 7:

 

 http://gee.cs.oswego.edu/dl/concurrency-interest/index.html

 

 You'll need jsr166y.jar in your classpath in order to use this

 library.  The basic idea is that Clojure collections, and most

 efficiently vectors, can be turned into parallel arrays for use by

 this library with the function par, although most of the functions

 take collections and will call par if needed, so normally you will

 only need to call par explicitly in order to attach bound/filter/map

 ops. Parallel arrays support the attachment of bounds, filters and

 mapping functions prior to realization/calculation, which happens as

 the result of any of several operations on the

 array (pvec/psort/pfilter-nils/pfilter-dupes). Rather than perform

 composite operations in steps, as would normally be done with

 sequences, maps and filters are instead attached and thus composed by

 providing ops to par. Note that there is an order sensitivity to the

 attachments - bounds precede filters precede mappings.  All operations

 then happen in parallel, using multiple threads and a sophisticated

 work-stealing system supported by fork-join, either when the array is

 realized, or to perform aggregate operations like preduce/pmin/pmax

 etc. A parallel array can be realized into a Clojure vector using

 pvec.

 ")

 

 (import '(jsr166y.forkjoin ParallelArray ParallelArrayWithBounds ParallelArrayWithFilter 

                            ParallelArrayWithMapping 

                            Ops$Op Ops$BinaryOp Ops$Reducer Ops$Predicate Ops$BinaryPredicate 

                            Ops$IntAndObjectPredicate Ops$IntAndObjectToObject))

 

 (defn- op [f]

   (proxy [Ops$Op] []

     (op [x] (f x))))

 

 (defn- binary-op [f]

   (proxy [Ops$BinaryOp] []

     (op [x y] (f x y))))

 

 (defn- int-and-object-to-object [f]

   (proxy [Ops$IntAndObjectToObject] []

     (op [i x] (f x i))))

 

 (defn- reducer [f]

   (proxy [Ops$Reducer] []

     (op [x y] (f x y))))

 

 (defn- predicate [f]

   (proxy [Ops$Predicate] []

     (op [x] (boolean (f x)))))

 

 (defn- binary-predicate [f]

   (proxy [Ops$BinaryPredicate] []

     (op [x y] (boolean (f x y)))))

 

 (defn- int-and-object-predicate [f]

   (proxy [Ops$IntAndObjectPredicate] []

     (op [i x] (boolean (f x i)))))

 

 (defn par

   "Creates a parallel array from coll. ops, if supplied, perform

   on-the-fly filtering or transformations during parallel realization

   or calculation. ops form a chain, and bounds must precede filters,

   must precede maps. ops must be a set of keyword value pairs of the

   following forms:

 

      :bound [start end] 

 

   Only elements from start (inclusive) to end (exclusive) will be

   processed when the array is realized.

 

      :filter pred 

 

   Filter preds remove elements from processing when the array is realized. pred

   must be a function of one argument whose return will be processed

   via boolean.

 

      :filter-index pred2 

 

   pred2 must be a function of two arguments, which will be an element

   of the collection and the corresponding index, whose return will be

   processed via boolean.

 

      :filter-with [pred2 coll2] 

 

   pred2 must be a function of two arguments, which will be

   corresponding elements of the 2 collections.

 

      :map f 

 

   Map fns will be used to transform elements when the array is

   realized. f must be a function of one argument.

 

      :map-index f2 

 

   f2 must be a function of two arguments, which will be an element of

   the collection and the corresponding index.

 

      :map-with [f2 coll2]

 

   f2 must be a function of two arguments, which will be corresponding

   elements of the 2 collections."

 

   ([coll] 

      (if (instance? ParallelArrayWithMapping coll)

        coll

        (. ParallelArray createUsingHandoff  

         (to-array coll) 

         (. ParallelArray defaultExecutor))))

   ([coll & ops]

      (reduce (fn [pa [op args]] 

                  (cond

                   (= op :bound) (. pa withBounds (args 0) (args 1))

                   (= op :filter) (. pa withFilter (predicate args))

                   (= op :filter-with) (. pa withFilter (binary-predicate (args 0)) (par (args 1)))

                   (= op :filter-index) (. pa withIndexedFilter (int-and-object-predicate args))

                   (= op :map) (. pa withMapping (parallel/op args))

                   (= op :map-with) (. pa withMapping (binary-op (args 0)) (par (args 1)))

                   (= op :map-index) (. pa withIndexedMapping (int-and-object-to-object args))

                   :else (throw (Exception. (str "Unsupported par op: " op)))))

              (par coll) 

              (partition 2 ops))))

 

 ;;;;;;;;;;;;;;;;;;;;; aggregate operations ;;;;;;;;;;;;;;;;;;;;;;

 (defn pany

   "Returns some (random) element of the coll if it satisfies the bound/filter/map"

   [coll] 

   (. (par coll) any))

 

 (defn pmax

   "Returns the maximum element, presuming Comparable elements, unless

   a Comparator comp is supplied"

   ([coll] (. (par coll) max))

   ([coll comp] (. (par coll) max comp)))

 

 (defn pmin

   "Returns the minimum element, presuming Comparable elements, unless

   a Comparator comp is supplied"

   ([coll] (. (par coll) min))

   ([coll comp] (. (par coll) min comp)))

 

 (defn- summary-map [s]

   {:min (.min s) :max (.max s) :size (.size s) :min-index (.indexOfMin s) :max-index (.indexOfMax s)})

 

 (defn psummary 

   "Returns a map of summary statistics (min. max, size, min-index, max-index, 

   presuming Comparable elements, unless a Comparator comp is supplied"

   ([coll] (summary-map (. (par coll) summary)))

   ([coll comp] (summary-map (. (par coll) summary comp))))

 

 (defn preduce 

   "Returns the reduction of the realized elements of coll

   using function f. Note f will not necessarily be called

   consecutively, and so must be commutative. Also note that 

   (f base an-element) might be performed many times, i.e. base is not

   an initial value as with sequential reduce."

   [f base coll]

   (. (par coll) (reduce (reducer f) base)))

 

 ;;;;;;;;;;;;;;;;;;;;; collection-producing operations ;;;;;;;;;;;;;;;;;;;;;;

 

 (defn- pa-to-vec [pa]

   (vec (. pa getArray)))

 

 (defn- pall

   "Realizes a copy of the coll as a parallel array, with any bounds/filters/maps applied"

   [coll]

   (if (instance? ParallelArrayWithMapping coll)

     (. coll all)

     (par coll)))

 

 (defn pvec 

   "Returns the realized contents of the parallel array pa as a Clojure vector"

   [pa] (pa-to-vec (pall pa)))

 

 (defn pdistinct

   "Returns a parallel array of the distinct elements of coll"

   [coll]

   (pa-to-vec (. (pall coll) allUniqueElements)))

 

 ;this doesn't work, passes null to reducer?

 (defn- pcumulate [coll f init]

   (.. (pall coll) (precumulate (reducer f) init)))

 

 (defn psort 

   "Returns a new vector consisting of the realized items in coll, sorted, 

   presuming Comparable elements, unless a Comparator comp is supplied"

   ([coll] (pa-to-vec (. (pall coll) sort)))

   ([coll comp] (pa-to-vec (. (pall coll) sort comp))))

 

 (defn pfilter-nils

   "Returns a vector containing the non-nil (realized) elements of coll"

   [coll]

   (pa-to-vec (. (pall coll) removeNulls)))

 

 (defn pfilter-dupes 

   "Returns a vector containing the (realized) elements of coll, 

   without any consecutive duplicates"

   [coll]

   (pa-to-vec (. (pall coll) removeConsecutiveDuplicates)))

 

 

 (comment

 (load-file "src/parallel.clj")

 (refer 'parallel)

 (pdistinct [1 2 3 2 1])

 ;(pcumulate [1 2 3 2 1] + 0) ;broken, not exposed

 (def a (make-array Object 1000000))

 (dotimes i (count a)

   (aset a i (rand-int i)))

 (time (reduce + 0 a))

 (time (preduce + 0 a))

 (time (count (distinct a)))

 (time (count (pdistinct a)))

 

 (preduce + 0 [1 2 3 2 1])

 (preduce + 0 (psort a))

 (pvec (par [11 2 3 2] :filter-index (fn [x i] (> i x))))

 (pvec (par [11 2 3 2] :filter-with [(fn [x y] (> y x)) [110 2 33 2]]))

 

 (psummary ;or pvec/pmax etc

  (par [11 2 3 2] 

       :filter-with [(fn [x y] (> y x)) 

                     [110 2 33 2]]

       :map #(* % 2)))

 

 (preduce + 0

   (par [11 2 3 2] 

        :filter-with [< [110 2 33 2]]))

 

 (time (reduce + 0 (map #(* % %) (range 1000000))))

 (time (preduce + 0 (par (range 1000000) :map-index *)))

 (def v (range 1000000))

 (time (preduce + 0 (par v :map-index *)))

 (time (preduce + 0 (par v :map  #(* % %))))

 (time (reduce + 0 (map #(* % %) v)))

 )