Adding immutable parallel hashsets.

Fixing an issue with hashset splitters where the splitting does not work
if some elements have already been iterated. Added parallel collections
exception handling. Added parallel collections break control. Renaming
ParHashTrie -> ParHashMap.

The part with immutable.{HashSet, HashMap} - review by rompf

1 files changed, 130 insertions, 58 deletions

diff --git a/src/library/scala/collection/immutable/HashSet.scala b/src/library/scala/collection/immutable/HashSet.scala
index 08e64d6709..3f04d63b40 100644
--- a/src/library/scala/collection/immutable/HashSet.scala
+++ b/src/library/scala/collection/immutable/HashSet.scala
@@ -14,6 +14,8 @@ package immutable
 import generic._
 import annotation.unchecked.uncheckedVariance
 
+import collection.parallel.immutable.ParHashSet
+
 /** This class implements immutable sets using a hash trie.
  *
  *  '''Note:''' the builder of a hash set returns specialized representations `EmptySet`,`Set1`,..., `Set4`
@@ -31,7 +33,9 @@ import annotation.unchecked.uncheckedVariance
 @serializable @SerialVersionUID(2L)
 class HashSet[A] extends Set[A]
                     with GenericSetTemplate[A, HashSet]
-                    with SetLike[A, HashSet[A]] {
+                    with SetLike[A, HashSet[A]]
+                    with Parallelizable[ParHashSet[A]]
+{
   override def companion: GenericCompanion[HashSet] = HashSet
 
   //class HashSet[A] extends Set[A] with SetLike[A, HashSet[A]] {
@@ -55,6 +59,8 @@ class HashSet[A] extends Set[A]
   def - (e: A): HashSet[A] =
     removed0(e, computeHash(e), 0)
 
+  def par = ParHashSet.fromTrie(this)
+
   protected def elemHashCode(key: A) = if (key == null) 0 else key.##
 
   protected final def improve(hcode: Int) = {
@@ -68,7 +74,7 @@ class HashSet[A] extends Set[A]
 
   protected def get0(key: A, hash: Int, level: Int): Boolean = false
 
-  protected def updated0(key: A, hash: Int, level: Int): HashSet[A] =
+  def updated0(key: A, hash: Int, level: Int): HashSet[A] =
     new HashSet.HashSet1(key, hash)
 
   protected def removed0(key: A, hash: Int, level: Int): HashSet[A] = this
@@ -169,7 +175,7 @@ object HashSet extends ImmutableSetFactory[HashSet] {
   }
 
 
-  class HashTrieSet[A](private var bitmap: Int, private var elems: Array[HashSet[A]],
+  class HashTrieSet[A](private var bitmap: Int, private[HashSet] var elems: Array[HashSet[A]],
       private var size0: Int) extends HashSet[A] {
 
     override def size = size0
@@ -239,60 +245,7 @@ object HashSet extends ImmutableSetFactory[HashSet] {
       }
     }
 
-
-    override def iterator = new Iterator[A] {
-      private[this] var depth = 0
-      private[this] var arrayStack = new Array[Array[HashSet[A]]](6)
-      private[this] var posStack = new Array[Int](6)
-
-      private[this] var arrayD = elems
-      private[this] var posD = 0
-
-      private[this] var subIter: Iterator[A] = null // to traverse collision nodes
-
-      def hasNext = (subIter ne null) || depth >= 0
-
-      def next: A = {
-        if (subIter ne null) {
-          val el = subIter.next
-          if (!subIter.hasNext)
-            subIter = null
-          el
-        } else
-          next0(arrayD, posD)
-      }
-
-      @scala.annotation.tailrec private[this] def next0(elems: Array[HashSet[A]], i: Int): A = {
-        if (i == elems.length-1) { // reached end of level, pop stack
-          depth -= 1
-          if (depth >= 0) {
-            arrayD = arrayStack(depth)
-            posD = posStack(depth)
-            arrayStack(depth) = null
-          } else {
-            arrayD = null
-            posD = 0
-          }
-        } else
-          posD += 1
-
-        elems(i) match {
-          case m: HashTrieSet[A] => // push current pos onto stack and descend
-            if (depth >= 0) {
-              arrayStack(depth) = arrayD
-              posStack(depth) = posD
-            }
-            depth += 1
-            arrayD = m.elems
-            posD = 0
-            next0(m.elems, 0)
-          case m: HashSet1[A] => m.key
-          case m =>
-            subIter = m.iterator
-            subIter.next
-        }
-      }
-    }
+    override def iterator = new TrieIterator[A](elems)
 
 /*
 
@@ -315,7 +268,6 @@ time { mNew.iterator.foreach( p => ()) }
 
 */
 
-
     override def foreach[U](f: A =>  U): Unit = {
       var i = 0;
       while (i < elems.length) {
@@ -325,6 +277,126 @@ time { mNew.iterator.foreach( p => ()) }
     }
   }
 
+
+  class TrieIterator[A](elems: Array[HashSet[A]]) extends Iterator[A] {
+    private[this] var depth = 0
+    private[this] var arrayStack = new Array[Array[HashSet[A]]](6)
+    private[this] var posStack = new Array[Int](6)
+
+    private[this] var arrayD = elems
+    private[this] var posD = 0
+
+    private[this] var subIter: Iterator[A] = null // to traverse collision nodes
+
+    def hasNext = (subIter ne null) || depth >= 0
+
+    def next: A = {
+      if (subIter ne null) {
+        val el = subIter.next
+        if (!subIter.hasNext)
+          subIter = null
+        el
+      } else
+        next0(arrayD, posD)
+    }
+
+    @scala.annotation.tailrec private[this] def next0(elems: Array[HashSet[A]], i: Int): A = {
+      if (i == elems.length-1) { // reached end of level, pop stack
+        depth -= 1
+        if (depth >= 0) {
+          arrayD = arrayStack(depth)
+          posD = posStack(depth)
+          arrayStack(depth) = null
+        } else {
+          arrayD = null
+          posD = 0
+        }
+      } else
+        posD += 1
+
+      elems(i) match {
+        case m: HashTrieSet[A] => // push current pos onto stack and descend
+          if (depth >= 0) {
+            arrayStack(depth) = arrayD
+            posStack(depth) = posD
+          }
+          depth += 1
+          arrayD = m.elems
+          posD = 0
+          next0(m.elems, 0)
+        case m: HashSet1[A] => m.key
+        case m =>
+          subIter = m.iterator
+          subIter.next
+      }
+    }
+
+    // assumption: contains 2 or more elements
+    // splits this iterator into 2 iterators
+    // returns the 1st iterator, its number of elements, and the second iterator
+    def split: ((Iterator[A], Int), Iterator[A]) = {
+      // 0) simple case: no elements have been iterated - simply divide arrayD
+      if (arrayD != null && depth == 0 && posD == 0) {
+        val (fst, snd) = arrayD.splitAt(arrayD.length / 2)
+        val szfst = fst.foldLeft(0)(_ + _.size)
+        return ((new TrieIterator(fst), szfst), new TrieIterator(snd))
+      }
+
+      // otherwise, some elements have been iterated over
+      // 1) collision case: if we have a subIter, we return subIter and elements after it
+      if (subIter ne null) {
+        val buff = subIter.toBuffer
+        subIter = null
+        ((buff.iterator, buff.length), this)
+      } else {
+        // otherwise find the topmost array stack element
+        if (depth > 0) {
+          // 2) topmost comes before (is not) arrayD
+          //    steal a portion of top to create a new iterator
+          val topmost = arrayStack(0)
+          if (posStack(0) == arrayStack(0).length - 1) {
+            // 2a) only a single entry left on top
+            // this means we have to modify this iterator - pop topmost
+            val snd = Array(arrayStack(0).last)
+            val szsnd = snd(0).size
+            // modify this - pop
+            depth -= 1
+            arrayStack = arrayStack.tail ++ Array[Array[HashSet[A]]](null)
+            posStack = posStack.tail ++ Array[Int](0)
+            // we know that `this` is not empty, since it had something on the arrayStack and arrayStack elements are always non-empty
+            ((new TrieIterator[A](snd), szsnd), this)
+          } else {
+            // 2b) more than a single entry left on top
+            val (fst, snd) = arrayStack(0).splitAt(arrayStack(0).length - (arrayStack(0).length - posStack(0) + 1) / 2)
+            arrayStack(0) = fst
+            val szsnd = snd.foldLeft(0)(_ + _.size)
+            ((new TrieIterator[A](snd), szsnd), this)
+          }
+        } else {
+          // 3) no topmost element (arrayD is at the top)
+          //    steal a portion of it and update this iterator
+          if (posD == arrayD.length - 1) {
+            // 3a) positioned at the last element of arrayD
+            val arr: Array[HashSet[A]] = arrayD(posD) match {
+              case c: HashSetCollision1[_] => c.asInstanceOf[HashSetCollision1[A]].ks.toList map { HashSet() + _ } toArray
+              case ht: HashTrieSet[_] => ht.asInstanceOf[HashTrieSet[A]].elems
+              case _ => error("cannot divide single element")
+            }
+            val (fst, snd) = arr.splitAt(arr.length / 2)
+            val szsnd = snd.foldLeft(0)(_ + _.size)
+            ((new TrieIterator(snd), szsnd), new TrieIterator(fst))
+          } else {
+            // 3b) arrayD has more free elements
+            val (fst, snd) = arrayD.splitAt(arrayD.length - (arrayD.length - posD + 1) / 2)
+            arrayD = fst
+            val szsnd = snd.foldLeft(0)(_ + _.size)
+            ((new TrieIterator[A](snd), szsnd), this)
+          }
+        }
+      }
+    }
+  }
+
   @serializable  @SerialVersionUID(2L) private class SerializationProxy[A,B](@transient private var orig: HashSet[A]) {
     private def writeObject(out: java.io.ObjectOutputStream) {
       val s = orig.size