added parallel fold

Author: richhickey

src/jvm/clojure/lang/PersistentHashMap.java

@@ -11,9 +11,8 @@
 package clojure.lang;
 
 import java.io.Serializable;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
+import java.util.concurrent.Callable;
 import java.util.concurrent.atomic.AtomicReference;
 
 /*
@@ -188,6 +187,22 @@ public Object kvreduce(IFn f, Object init){
     return init;
 }
 
+public Object fold(long n, final IFn combinef, final IFn reducef,
+                   IFn fjinvoke, final IFn fjtask, final IFn fjfork, final IFn fjjoin){
+	//we are ignoring n for now
+	Callable top = new Callable(){
+		public Object call() throws Exception{
+			Object ret = combinef.invoke();
+			if(root != null)
+				ret = combinef.invoke(ret, root.fold(combinef,reducef,fjtask,fjfork,fjjoin));
+			return hasNull?
+			       combinef.invoke(ret,reducef.invoke(combinef.invoke(),null,nullValue))
+			       :ret;
+		}
+	};
+	return fjinvoke.invoke(top);
+}
+
 public int count(){
 	return count;
 }
@@ -324,6 +339,7 @@ static interface INode extends Serializable {
 
     public Object kvreduce(IFn f, Object init);
 
+	Object fold(IFn combinef, IFn reducef, IFn fjtask, IFn fjfork, IFn fjjoin);
 }
 
 final static class ArrayNode implements INode{
@@ -395,6 +411,52 @@ public Object kvreduce(IFn f, Object init){
         return init;
     }
 
+	public Object fold(final IFn combinef, final IFn reducef,
+	                   final IFn fjtask, final IFn fjfork, final IFn fjjoin){
+		List<Callable> tasks = new ArrayList();
+		for(final INode node : array){
+			if(node != null){
+				tasks.add(new Callable(){
+					public Object call() throws Exception{
+						return node.fold(combinef, reducef, fjtask, fjfork, fjjoin);
+					}
+				});
+				}
+			}
+
+		return foldTasks(tasks,combinef,fjtask,fjfork,fjjoin);
+		}
+
+	static public Object foldTasks(List<Callable> tasks, final IFn combinef,
+	                          final IFn fjtask, final IFn fjfork, final IFn fjjoin){
+
+		if(tasks.isEmpty())
+			return combinef.invoke();
+
+		if(tasks.size() == 1){
+			Object ret = null;
+			try
+				{
+				return tasks.get(0).call();
+				}
+			catch(Exception e)
+				{
+				//aargh
+				}
+			}
+
+		List<Callable> t1 = tasks.subList(0,tasks.size()/2);
+		final List<Callable> t2 = tasks.subList(tasks.size()/2, tasks.size());
+
+		Object forked = fjfork.invoke(fjtask.invoke(new Callable() {
+			public Object call() throws Exception{
+				return foldTasks(t2,combinef,fjtask,fjfork,fjjoin);
+			}
+		}));
+
+		return combinef.invoke(foldTasks(t1,combinef,fjtask,fjfork,fjjoin),fjjoin.invoke(forked));
+	}
+
 
 	private ArrayNode ensureEditable(AtomicReference<Thread> edit){
 		if(this.edit == edit)
@@ -629,6 +691,9 @@ public Object kvreduce(IFn f, Object init){
          return NodeSeq.kvreduce(array,f,init);
     }
 
+	public Object fold(IFn combinef, IFn reducef, IFn fjtask, IFn fjfork, IFn fjjoin){
+		return NodeSeq.kvreduce(array, reducef, combinef.invoke());
+	}
 
 	private BitmapIndexedNode ensureEditable(AtomicReference<Thread> edit){
 		if(this.edit == edit)
@@ -818,6 +883,10 @@ public Object kvreduce(IFn f, Object init){
          return NodeSeq.kvreduce(array,f,init);
     }
 
+	public Object fold(IFn combinef, IFn reducef, IFn fjtask, IFn fjfork, IFn fjjoin){
+		return NodeSeq.kvreduce(array, reducef, combinef.invoke());
+	}
+
 	public int findIndex(Object key){
 		for(int i = 0; i < 2*count; i+=2)
 			{