Adding primary version of parallel hash tries.

No review.

5 files changed, 99 insertions, 38 deletions

diff --git a/src/library/scala/collection/immutable/HashMap.scala b/src/library/scala/collection/immutable/HashMap.scala
index 11292bdf0c..57fc72d8c7 100644
--- a/src/library/scala/collection/immutable/HashMap.scala
+++ b/src/library/scala/collection/immutable/HashMap.scala
@@ -105,10 +105,9 @@ object HashMap extends ImmutableMapFactory[HashMap] {
   // TODO: add HashMap2, HashMap3, ...
 
   // statistics - will remove in future
-  var dives = 0
-  var colls = 0
-  var two_colls = 0
-  var two_nocolls = 0
+  var bothsingle = 0
+  var bothtries = 0
+  var onetrie = 0
 
 
   class HashMap1[A,+B](private[HashMap] var key: A, private[HashMap] var hash: Int, private[HashMap] var value: (B @uncheckedVariance), private[HashMap] var kv: (A,B @uncheckedVariance)) extends HashMap[A,B] {
@@ -171,7 +170,11 @@ object HashMap extends ImmutableMapFactory[HashMap] {
     override def iterator: Iterator[(A,B)] = Iterator(ensurePair)
     override def foreach[U](f: ((A, B)) => U): Unit = f(ensurePair)
     private[HashMap] def ensurePair: (A,B) = if (kv ne null) kv else { kv = (key, value); kv }
-    protected override def combine0[B1 >: B](that: HashMap[A, B1], level: Int): HashMap[A, B1] = that.updated0(key, hash, level, value, kv)
+    protected override def combine0[B1 >: B](that: HashMap[A, B1], level: Int): HashMap[A, B1] = {
+      // if (that.isInstanceOf[HashMap1[_, _]]) bothsingle += 1
+      // else onetrie += 1
+      that.updated0(key, hash, level, value, kv)
+    }
   }
 
   private class HashMapCollision1[A,+B](private[HashMap] var hash: Int, var kvs: ListMap[A,B @uncheckedVariance]) extends HashMap[A,B] {
@@ -266,8 +269,7 @@ object HashMap extends ImmutableMapFactory[HashMap] {
         Array.copy(elems, 0, elemsNew, 0, offset)
         elemsNew(offset) = new HashMap1(key, hash, value, kv)
         Array.copy(elems, offset, elemsNew, offset + 1, elems.length - offset)
-        val bitmapNew = bitmap | mask
-        new HashTrieMap(bitmapNew, elemsNew, size + 1)
+        new HashTrieMap(bitmap | mask, elemsNew, size + 1)
       }
     }
 
@@ -427,7 +429,7 @@ time { mNew.iterator.foreach( p => ()) }
       i
     }
 
-    override def split: Seq[HashMap[A, B]] = {
+    override def split: Seq[HashMap[A, B]] = if (size == 1) Seq(this) else {
       // printBitmap(bitmap)
       // println(elems.toList)
 
@@ -449,12 +451,10 @@ time { mNew.iterator.foreach( p => ()) }
 
     protected override def combine0[B1 >: B](that: HashMap[A, B1], level: Int): HashMap[A, B1] = that match {
       case hm: HashMap1[_, _] =>
+        // onetrie += 1
         this.updated0(hm.key, hm.hash, level, hm.value.asInstanceOf[B1], hm.kv)
-      case hm: HashMapCollision1[_, _] =>
-        var m: HashMap[A, B1] = this
-        for (p <- that) m = m.updated0(p._1, computeHash(p._1), level, p._2, p)
-        m
       case hm: HashTrieMap[_, _] =>
+        // bothtries += 1
         val that = hm.asInstanceOf[HashTrieMap[A, B1]]
         val thiselems = this.elems
         val thatelems = that.elems
@@ -465,7 +465,7 @@ time { mNew.iterator.foreach( p => ()) }
 	val subcount = Integer.bitCount(thisbm | thatbm)
 
         // construct a new array of appropriate size
-        val combined = new Array[HashMap[A, B1 @uncheckedVariance]](subcount)
+        val combined = new Array[HashMap[A, B1]](subcount)
 
 	// run through both bitmaps and add elements to it
         var i = 0
@@ -497,7 +497,7 @@ time { mNew.iterator.foreach( p => ()) }
             // and compare a and b defined as below:
             val a = thislsb - 1
             val b = thatlsb - 1
-            //  ! our case indeed is more specific, but this didn't help:
+            // ! our case indeed is more specific, but this didn't help:
             // if ((thislsb > 0 && thislsb < thatlsb) || thatlsb == 0 || (thatlsb < 0 && thislsb != 0)) {
             if ((a < b) ^ (a < 0) ^ (b < 0)) {
               // println("an element from this trie")
@@ -518,16 +518,8 @@ time { mNew.iterator.foreach( p => ()) }
           i += 1
         }
 
-        val res = new HashTrieMap[A, B1](this.bitmap | that.bitmap, combined, totalelems)
-        // if (!check(this, that, res)) { TODO remove
-        //   printBitmap(this.bitmap)
-        //   printBitmap(that.bitmap)
-        //   printBitmap(res.bitmap)
-        //   println(this.bitmap)
-        //   System.exit(1)
-        // }
-        res
-      case empty: HashMap[_, _] => this
+        new HashTrieMap[A, B1](this.bitmap | that.bitmap, combined, totalelems)
+      case hm: HashMapCollision1[_, _] => that.combine0(this, level)
       case _ => error("section supposed to be unreachable.")
     }
 
diff --git a/src/parallel-collections/scala/collection/parallel/ParallelMap.scala b/src/parallel-collections/scala/collection/parallel/ParallelMap.scala
index c03f14b198..2eaf474b07 100644
--- a/src/parallel-collections/scala/collection/parallel/ParallelMap.scala
+++ b/src/parallel-collections/scala/collection/parallel/ParallelMap.scala
@@ -20,16 +20,19 @@ extends Map[K, V]
    with ParallelMapLike[K, V, ParallelMap[K, V], Map[K, V]]
 { self =>
 
-  override def empty: ParallelMap[K, V] = null // TODO
+  override def empty: ParallelMap[K, V] = new immutable.ParallelHashTrie[K, V]
+
+  override def stringPrefix = "ParallelMap"
 
 }
 
 
 
 object ParallelMap extends ParallelMapFactory[ParallelMap] {
-  def empty[A, B]: ParallelMap[A, B] = null // TODO
+  def empty[K, V]: ParallelMap[K, V] = new immutable.ParallelHashTrie[K, V]
+
+  implicit def canBuildFrom[K, V]: CanBuildFromParallel[Coll, (K, V), ParallelMap[K, V]] = new ParallelMapCanBuildFrom[K, V]
 
-  implicit def canBuildFrom[A, B]: CanBuildFromParallel[Coll, (A, B), Map[A, B]] = null // TODO
 }
 
 
diff --git a/src/parallel-collections/scala/collection/parallel/immutable/ParallelHashTrie.scala b/src/parallel-collections/scala/collection/parallel/immutable/ParallelHashTrie.scala
index be379d9e5e..577fb2642b 100644
--- a/src/parallel-collections/scala/collection/parallel/immutable/ParallelHashTrie.scala
+++ b/src/parallel-collections/scala/collection/parallel/immutable/ParallelHashTrie.scala
@@ -8,6 +8,10 @@ package scala.collection.parallel.immutable
 
 import scala.collection.parallel.ParallelMap
 import scala.collection.parallel.ParallelMapLike
+import scala.collection.parallel.Combiner
+import scala.collection.parallel.EnvironmentPassingCombiner
+import scala.collection.generic.ParallelMapFactory
+import scala.collection.generic.CanBuildFromParallel
 import scala.collection.immutable.HashMap
 
 
@@ -15,33 +19,85 @@ import scala.collection.immutable.HashMap
 
 
 
-
-
-class ParallelHashTrie[K, +V]
+class ParallelHashTrie[K, +V] private[immutable] (private[this] val trie: HashMap[K, V])
 extends ParallelMap[K, V]
    with ParallelMapLike[K, V, ParallelHashTrie[K, V], HashMap[K, V]]
 { self =>
 
-  override def empty: ParallelHashTrie[K, V] = null // TODO
+  def this() = this(HashMap.empty[K, V])
+
+  override def empty: ParallelHashTrie[K, V] = new ParallelHashTrie[K, V]
+
+  def parallelIterator = new ParallelHashTrieIterator(trie) with SCPI
 
-  def parallelIterator = null // TODO
+  def seq = trie
 
-  def seq = null // TODO
+  def -(k: K) = new ParallelHashTrie(trie - k)
 
-  def -(k: K) = null // TODO
+  def +[U >: V](kv: (K, U)) = new ParallelHashTrie(trie + kv)
 
-  def +[U >: V](kv: (K, U)) = null // TODO
+  def get(k: K) = trie.get(k)
 
-  def get(k: K) = None // TODO
+  override def size = trie.size
+
+  type SCPI = SignalContextPassingIterator[ParallelHashTrieIterator]
+
+  class ParallelHashTrieIterator(trie: HashMap[K, V])
+  extends super.ParallelIterator {
+  self: SignalContextPassingIterator[ParallelHashTrieIterator] =>
+    println("created iterator")
+    var i = 0
+    lazy val triter = trie.iterator
+    def split: Seq[ParallelIterator] = {
+      println("splitting " + trie + " into " + (trie.split.map(_.size)))
+      trie.split.map(new ParallelHashTrieIterator(_) with SCPI)
+    }
+    def next: (K, V) = {
+      println("taking next after " + i)
+      i += 1
+      triter.next
+    }
+    def hasNext: Boolean = i < trie.size
+    def remaining = trie.size - i
+  }
 
 }
 
 
+object ParallelHashTrie extends ParallelMapFactory[ParallelHashTrie] {
+  def empty[K, V]: ParallelHashTrie[K, V] = new ParallelHashTrie[K, V]
+
+  implicit def canBuildFrom[K, V]: CanBuildFromParallel[ParallelHashTrie[K, V], (K, V), ParallelMap[K, V]] = new ParallelMapCanBuildFrom[K, V]
+}
 
 
+trait HashTrieCombiner[K, V]
+extends Combiner[(K, V), ParallelHashTrie[K, V]] {
+self: EnvironmentPassingCombiner[(K, V), ParallelHashTrie[K, V]] =>
+  private var trie: HashMap[K, V] = HashMap.empty[K, V]
+
+  def size: Int = trie.size
+
+  def clear = trie = HashMap.empty[K, V]
+
+  def +=(elem: (K, V)) = { trie += elem; this }
+
+  def result = new ParallelHashTrie[K, V](trie)
+
+  def combine[N <: (K, V), NewTo >: ParallelHashTrie[K, V]](other: Combiner[N, NewTo]): Combiner[N, NewTo] = {
+    if (other.isInstanceOf[HashTrieCombiner[_, _]]) {
+      val that = other.asInstanceOf[HashTrieCombiner[K, V]]
+      val ncombiner = HashTrieCombiner[K, V]
+      ncombiner.trie = this.trie combine that.trie
+      ncombiner
+    } else error("Unexpected combiner type.")
+  }
+
+}
 
-object ParallelHashTrie {
 
+object HashTrieCombiner {
+  def apply[K, V] = new HashTrieCombiner[K, V] with EnvironmentPassingCombiner[(K, V), ParallelHashTrie[K, V]] {}
 }
 
 
diff --git a/test/benchmarks/src/scala/collection/parallel/Benchmarking.scala b/test/benchmarks/src/scala/collection/parallel/Benchmarking.scala
index f6aa63cb1a..c52724757a 100644
--- a/test/benchmarks/src/scala/collection/parallel/Benchmarking.scala
+++ b/test/benchmarks/src/scala/collection/parallel/Benchmarking.scala
@@ -93,6 +93,7 @@ trait BenchmarkRegister {
   register(hashtries.Construct)
   register(hashtries.Lookup)
   register(hashtries.Combine)
+  register(hashtries.MultipleCombine)
 }
 
 
diff --git a/test/benchmarks/src/scala/collection/parallel/benchmarks/hashtries/Combine.scala b/test/benchmarks/src/scala/collection/parallel/benchmarks/hashtries/Combine.scala
index f04688c7f9..16f791a710 100644
--- a/test/benchmarks/src/scala/collection/parallel/benchmarks/hashtries/Combine.scala
+++ b/test/benchmarks/src/scala/collection/parallel/benchmarks/hashtries/Combine.scala
@@ -22,13 +22,22 @@ class Combine(val size: Int, val parallelism: Int, val runWhat: String) extends
   def runseq = runhashtrie
   def runhashtrie = {
     hashtrie combine thattrie
+    // println
+    // println("both tries:         " + HashTrie.bothtries)
+    // println("one trie, one item: " + HashTrie.onetrie)
+    // println("both single:        " + HashTrie.bothsingle)
+    // System exit 1
+  }
+  def rundestructive = {
+    hashtrie combine thattrie
   }
   def runappendtrie = hashtrie ++ thattrie
   def runhashmap = hashmap ++ thatmap
   def companion = Combine
-  def comparisonMap = Map("hashtrie" -> runhashtrie _, "hashmap" -> runhashmap _, "appendtrie" -> runappendtrie _)
+  def comparisonMap = Map("hashtrie" -> runhashtrie _, "hashmap" -> runhashmap _, "destruct" -> rundestructive _, "appendtrie" -> runappendtrie _)
   override def reset = runWhat match {
     case "appendtrie" => initHashTrie
+    case "destruct" => initHashTrie
     case _ => super.reset
   }
 }