parallel.clj, interface to Fork/Join framework

3 files changed, 122 insertions, 33 deletions

diff --git a/src/boot.clj b/src/boot.clj
index 3486d92a..bc8f916f 100644
--- a/src/boot.clj
+++ b/src/boot.clj
@@ -1569,10 +1569,7 @@ not-every? (comp not every?))
 (defn to-array
   "Returns an array of Objects containing the contents of coll, which
   can be any Collection.  Maps to java.util.Collection.toArray()."
-  [#^java.util.Collection coll]
-    (if (zero? (count coll))
-      (. clojure.lang.RT EMPTY_ARRAY)
-      (. coll (toArray))))
+  [coll] (. clojure.lang.RT (toArray coll)))
 
 (defn to-array-2d
   "Returns a (potentially-ragged) 2-dimensional array of Objects
diff --git a/src/jvm/clojure/lang/RT.java b/src/jvm/clojure/lang/RT.java
index fb46959e..eb11c5c4 100644
--- a/src/jvm/clojure/lang/RT.java
+++ b/src/jvm/clojure/lang/RT.java
@@ -203,7 +203,7 @@ static AtomicInteger id = new AtomicInteger(1);
 static final public DynamicClassLoader ROOT_CLASSLOADER = new DynamicClassLoader();
 
 static public void addURL(Object url) throws Exception{
-	URL u = (url instanceof String)?(new URL((String) url)):(URL)url;
+	URL u = (url instanceof String) ? (new URL((String) url)) : (URL) url;
 	ROOT_CLASSLOADER.addURL(u);
 }
 
@@ -285,11 +285,11 @@ static public void init() throws Exception{
 }
 
 static void doInit() throws Exception{
-	loadResourceScript(RT.class,"boot.clj");
-	loadResourceScript(RT.class,"proxy.clj");
-	loadResourceScript(RT.class,"zip.clj");
-	loadResourceScript(RT.class,"xml.clj");
-	loadResourceScript(RT.class,"set.clj");
+	loadResourceScript(RT.class, "boot.clj");
+	loadResourceScript(RT.class, "proxy.clj");
+	loadResourceScript(RT.class, "zip.clj");
+	loadResourceScript(RT.class, "xml.clj");
+	loadResourceScript(RT.class, "set.clj");
 
 	Var.pushThreadBindings(
 			RT.map(CURRENT_NS, CURRENT_NS.get(),
@@ -303,7 +303,7 @@ static void doInit() throws Exception{
 		Var refer = var("clojure", "refer");
 		in_ns.invoke(USER);
 		refer.invoke(CLOJURE);
-		loadResourceScript(RT.class,"user.clj");
+		loadResourceScript(RT.class, "user.clj");
 		}
 	finally
 		{
@@ -790,6 +790,25 @@ static public ISeq arrayToList(Object[] a) throws Exception{
 	return ret;
 }
 
+static public Object[] toArray(Object coll) throws Exception{
+	if(coll == null)
+		return EMPTY_ARRAY;
+	else if(coll instanceof Object[])
+		return (Object[]) coll;
+	else if(coll instanceof Collection)
+		return ((Collection) coll).toArray();
+	else if(coll instanceof String)
+		{
+		char[] chars = ((String) coll).toCharArray();
+		Object[] ret = new Object[chars.length];
+		for(int i = 0; i < chars.length; i++)
+			ret[i] = chars[i];
+		return ret;
+		}
+	else
+		throw new Exception("Unable to convert: " + coll.getClass() + " to Object[]");
+}
+
 static public Object[] seqToArray(ISeq seq){
 	int len = length(seq);
 	Object[] ret = new Object[len];
diff --git a/src/parallel.clj b/src/parallel.clj
index f2896eba..8f99d0e6 100644
--- a/src/parallel.clj
+++ b/src/parallel.clj
@@ -10,24 +10,54 @@
 (clojure/refer 'clojure)
 
 (import '(jsr166y.forkjoin ParallelArray ParallelArrayWithBounds ParallelArrayWithFilter 
-                           ParallelArrayWithMapping Ops$Reducer Ops$Predicate))
+                           ParallelArrayWithMapping 
+                           Ops$Op Ops$BinaryOp Ops$Reducer Ops$Predicate Ops$BinaryPredicate Ops$IntAndObjectPredicate
+                           Ops$IntAndObjectToObject))
 
-(defn parray [coll]
+(defn par [coll]
   (if (instance? ParallelArrayWithMapping coll)
     coll
-    (ParallelArray.createUsingHandoff 
-     (if (identical? (class coll) (class (clojure.lang.RT.EMPTY_ARRAY)))
-       coll 
-       (to-array coll)) 
+    (ParallelArray.createUsingHandoff  
+     (to-array coll) 
      (ParallelArray.defaultExecutor))))
 
-(defn- pall [coll]
+(defn pall [coll]
   (if (instance? ParallelArrayWithMapping coll)
     (. coll all)
-    (parray coll)))
+    (par coll)))
+
+(defn pany [coll] 
+  (. (par coll) any))
+
+(defn pmax 
+  ([coll] (. (par coll) max))
+  ([coll comp] (. (par coll) max comp)))
+
+(defn pmin
+  ([coll] (. (par coll) min))
+  ([coll comp] (. (par coll) min comp)))
+
+(defn- summary-map [s]
+  {:min (s.min) :max (s.max) :size (s.size) :min-index (s.indexOfMin) :max-index (s.indexOfMax)})
+
+(defn psummary 
+  ([coll] (summary-map (. (par coll) summary)))
+  ([coll comp] (summary-map (. (par coll) summary comp))))
 
 (defn punique [coll]
-  (. (parray coll) allUniqueElements))
+  (. (par coll) allUniqueElements))
+
+(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 i x))))
 
 (defn- reducer [f]
   (proxy [Ops$Reducer] []
@@ -37,31 +67,61 @@
   (proxy [Ops$Predicate] []
     (op [x] (boolean (f x)))))
 
-;this doesn't work, passes null to reducer
+(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 i x)))))
+
+;this doesn't work, passes null to reducer?
 (defn pcumulate [coll f init]
-  (.. (pall coll) (cumulate (reducer f) init)))
+  (.. (pall coll) (precumulate (reducer f) init)))
+
+(defn preduce [f base coll]
+  (. (par coll) (reduce (reducer f) base)))
+
+(defn psort 
+  ([coll] (. (pall coll) sort))
+  ([coll comp] (. (pall coll) sort comp)))
+
+(defn pfilter-nils [coll]
+  (. (pall coll) removeNulls))
+
+(defn pfilter-dupes [coll]
+  (. (pall coll) removeConsecutiveDuplicates))
 
-(defn preduce [f init coll]
-  (. (parray coll) (reduce (reducer f) init)))
+(defn with-bounds [coll start end]
+  (. (par coll) withBounds start end))
 
-(defn psort [coll]
-  (. (pall coll) sort))
+(defn with-filter
+  ([coll f]
+     (. (par coll) withFilter (predicate f)))
+  ([coll f2 coll2]
+     (. (par coll) withFilter (binary-predicate f2) (par coll2))))
 
-(defn pbound [coll start end]
-  (. (parray coll) withBounds start end))
+(defn with-indexed-filter
+  ([coll f]
+     (. (par coll) withIndexedFilter (int-and-object-predicate f))))
 
-(defn pfilter
+(defn with-mapping
   ([coll f]
-     (. (parray coll) withFilter (predicate f))))
+     (. (par coll) withMapping (op f)))
+  ([coll f2 coll2]
+     (. (par coll) withMapping (binary-op f2) (par coll2))))
 
-(defn pfilter
+(defn with-indexed-mapping
   ([coll f]
-     (. (parray coll) withFilter (predicate f))))
+     (. (par coll) withIndexedMapping (int-and-object-to-object f))))
+
+(defn pvec [pa]
+  (vec (. (pall pa) getArray)))
 
 (comment
 
 (punique [1 2 3 2 1])
-(def v (range 10000000))
+(pcumulate [1 2 3 2 1] + 0) ;broken
 (def a (make-array Object 1000000))
 (dotimes i (count a)
   (aset a i i))
@@ -69,4 +129,17 @@
 (time (preduce + 0 a))
 (preduce + 0 [1 2 3 2 1])
 (preduce + 0 (psort a))
+(pall (with-indexed-filter (par [11 2 3 2]) (fn [i x] (> i x))))
+(pall (with-filter (par [11 2 3 2]) (fn [x y] (> y x)) (par [110 2 33 2])))
+
+(psummary ;or pvec/pmax etc
+ (-> (par [11 2 3 2]) 
+     (with-filter (fn [x y] (> y x)) 
+                  (par [110 2 33 2]))
+     (with-mapping #(* % 2))))
+
+(preduce + 0
+  (-> (par [11 2 3 2]) 
+      (with-filter (fn [x y] (> y x)) 
+                   (par [110 2 33 2]))))
 )
\ No newline at end of file