Reimplementing parallel views to solve several ...

Reimplementing parallel views to solve several performance glitches. No
review.

7 files changed, 157 insertions, 69 deletions

diff --git a/test/benchmarks/src/scala/collection/parallel/Benchmarking.scala b/test/benchmarks/src/scala/collection/parallel/Benchmarking.scala
index 2cc3abaf27..f733f4154b 100644
--- a/test/benchmarks/src/scala/collection/parallel/Benchmarking.scala
+++ b/test/benchmarks/src/scala/collection/parallel/Benchmarking.scala
@@ -72,6 +72,10 @@ trait BenchmarkRegister {
   register(parallel_view.DummyViewBenchList.MediumReduce)
   register(parallel_view.DummyViewBenchList.ModifyThenReduce)
   register(parallel_view.DummyViewBenchList.ModifyThenForce)
+  register(parallel_view.DummyViewBenchList.Iteration)
+  register(parallel_view.DummyViewBenchList.IterationS)
+  register(parallel_view.DummyViewBenchList.IterationM)
+  register(parallel_view.DummyViewBenchList.IterationA)
 
   // parallel ranges
   register(parallel_range.RangeBenches.Reduce)
diff --git a/test/benchmarks/src/scala/collection/parallel/benchmarks/generic/Dummy.scala b/test/benchmarks/src/scala/collection/parallel/benchmarks/generic/Dummy.scala
index 2b2ad81af6..ae93c7adf4 100644
--- a/test/benchmarks/src/scala/collection/parallel/benchmarks/generic/Dummy.scala
+++ b/test/benchmarks/src/scala/collection/parallel/benchmarks/generic/Dummy.scala
@@ -7,6 +7,8 @@ class Dummy(val in: Int) {
   var num = in
   override def toString = in.toString
   override def hashCode = in
+  def dummy = num + in
+  def one = "1".length
 }
 
 
@@ -37,7 +39,7 @@ object DummyOperators extends Operators[Dummy] {
     a.in % 2 == 0
   }
   val mapper = (a: Dummy) => {
-    a.num = a.in % 2
+    a.num = a.dummy + a.num + a.in + a.one
     a
   }
   val heavymapper = (a: Dummy) => {
@@ -51,9 +53,16 @@ object DummyOperators extends Operators[Dummy] {
   val taker = (a: Dummy) => {
     a.in >= 0
   }
+  val eachFun: Dummy => Unit = (d: Dummy) => {
+    d.dummy
+  }
+  override def sequence(sz: Int): Seq[Dummy] = {
+    val pa = new collection.parallel.mutable.ParArray[Dummy](sz)
+    for (i <- 0 until sz) pa(i) = new Dummy(i)
+    pa
+  }
 }
 
 
 
 
-
diff --git a/test/benchmarks/src/scala/collection/parallel/benchmarks/generic/Operators.scala b/test/benchmarks/src/scala/collection/parallel/benchmarks/generic/Operators.scala
index 1268f94bac..4bd693a933 100644
--- a/test/benchmarks/src/scala/collection/parallel/benchmarks/generic/Operators.scala
+++ b/test/benchmarks/src/scala/collection/parallel/benchmarks/generic/Operators.scala
@@ -14,6 +14,8 @@ trait Operators[T] {
   def mapper2: T => T = error("unsupported")
   def heavymapper: T => T
   def taker: T => Boolean
+  def eachFun: T => Unit
+  def sequence(sz: Int): Seq[T] = error("unsupported")
 
 }
 
@@ -40,6 +42,9 @@ trait IntOperators extends Operators[Int] {
     n + sum
   }
   val taker: Int => Boolean = _ < 10000
+  val eachFun: Int => Unit = { n =>
+    n % 2 == 0
+  }
 
 }
 
diff --git a/test/benchmarks/src/scala/collection/parallel/benchmarks/generic/ParallelBenches.scala b/test/benchmarks/src/scala/collection/parallel/benchmarks/generic/ParallelBenches.scala
index da5a2a63ca..fd4e87ab4c 100644
--- a/test/benchmarks/src/scala/collection/parallel/benchmarks/generic/ParallelBenches.scala
+++ b/test/benchmarks/src/scala/collection/parallel/benchmarks/generic/ParallelBenches.scala
@@ -4,55 +4,56 @@ package generic
 
 
 
+import scala.collection.SeqView
 
 
 
-
-trait ParIterableBench[T, Coll <: ParIterable[T]] extends collection.parallel.benchmarks.Bench {
+trait ParIterableBenches[T, Coll <: ParIterable[T]] {
 self =>
 
-  protected var seqcoll: Iterable[T] = null
-  protected var parcoll: Coll = null.asInstanceOf[Coll]
-
-  reset
-
-  def reset = runWhat match {
-    case "seq" => this.seqcoll = createSequential(size, parallelism)
-    case "par" => this.parcoll = createParallel(size, parallelism)
-    case _ =>
-  }
-
-  def nameOfCollection: String
-  def operators: Operators[T]
   def createSequential(sz: Int, p: Int): Iterable[T]
   def createParallel(sz: Int, p: Int): Coll
+  def nameOfCollection: String
+  def operators: Operators[T]
 
   trait IterableBenchCompanion extends BenchCompanion {
     def collectionName = self.nameOfCollection
   }
 
-  trait IterableBench extends ParIterableBench[T, Coll] {
+  trait IterableBench extends collection.parallel.benchmarks.Bench {
+    protected var seqcoll: Iterable[T] = null
+    protected var parcoll: Coll = null.asInstanceOf[Coll]
+
+    reset
+
+    def reset = runWhat match {
+      case "seq" => this.seqcoll = createSequential(size, parallelism)
+      case "par" => this.parcoll = createParallel(size, parallelism)
+      case _ =>
+    }
+
     def nameOfCollection = self.nameOfCollection
     def operators = self.operators
     def createSequential(sz: Int, p: Int) = self.createSequential(size, parallelism)
     def createParallel(sz: Int, p: Int) = self.createParallel(size, parallelism)
     def forkJoinPool: scala.concurrent.forkjoin.ForkJoinPool = self.forkJoinPool
-  }
 
-  def forkJoinPool: scala.concurrent.forkjoin.ForkJoinPool
+    override def printResults {
+      println(" --- Fork join pool state --- ")
+      println("Parallelism: " + forkJoinPool.getParallelism)
+      println("Active threads: " + forkJoinPool.getActiveThreadCount)
+      println("Work stealings: "  + forkJoinPool.getStealCount)
+    }
 
-  override def printResults {
-    println(" --- Fork join pool state --- ")
-    println("Parallelism: " + forkJoinPool.getParallelism)
-    println("Active threads: " + forkJoinPool.getActiveThreadCount)
-    println("Work stealings: "  + forkJoinPool.getStealCount)
   }
 
+  def forkJoinPool: scala.concurrent.forkjoin.ForkJoinPool
+
 }
 
 
-trait ParSeqBench[T, Coll <: ParSeq[T]] extends ParIterableBench[T, Coll] {
-  self =>
+trait ParSeqBenches[T, Coll <: ParSeq[T]] extends ParIterableBenches[T, Coll] {
+self =>
 
   def createSequential(sz: Int, p: Int): Seq[T]
 
@@ -60,27 +61,18 @@ trait ParSeqBench[T, Coll <: ParSeq[T]] extends ParIterableBench[T, Coll] {
     def collectionName = self.nameOfCollection
   }
 
-  trait SeqBench extends IterableBench with ParSeqBench[T, Coll] {
-    override def createSequential(sz: Int, p: Int) = self.createSequential(size, parallelism)
+  trait SeqBench extends IterableBench {
+    override def createSequential(sz: Int, p: Int) = self.createSequential(sz, p)
   }
 
 }
 
 
-trait NotBenchmark {
-  lazy val runWhat = "";
-  val size = -1
-  val parallelism = -1
-  def runpar {}
-  def runseq {}
-  def companion = throw new UnsupportedOperationException
-}
 
 
-/**
- * Standard benchmarks for collections.
+/** Standard benchmarks for collections.
  */
-trait StandardParIterableBench[T, Coll <: ParIterable[T]] extends ParIterableBench[T, Coll] {
+trait StandardParIterableBenches[T, Coll <: ParIterable[T]] extends ParIterableBenches[T, Coll] {
 
   object Reduce extends IterableBenchCompanion {
     override def defaultSize = 50000
@@ -89,7 +81,7 @@ trait StandardParIterableBench[T, Coll <: ParIterable[T]] extends ParIterableBen
   }
 
   class Reduce(val size: Int, val parallelism: Int, val runWhat: String)
-  extends IterableBench with StandardParIterableBench[T, Coll] {
+  extends IterableBench {
     def comparisonMap = collection.Map()
     def runseq = this.seqcoll.reduceLeft(operators.reducer)
     def runpar = this.parcoll.reduce(operators.reducer)
@@ -103,7 +95,7 @@ trait StandardParIterableBench[T, Coll <: ParIterable[T]] extends ParIterableBen
   }
 
   class ReduceMedium(val size: Int, val parallelism: Int, val runWhat: String)
-  extends IterableBench with StandardParIterableBench[T, Coll] {
+  extends IterableBench {
     def comparisonMap = collection.Map()
     def runseq = this.seqcoll.reduceLeft(operators.mediumreducer)
     def runpar = this.parcoll.reduce(operators.mediumreducer)
@@ -117,7 +109,7 @@ trait StandardParIterableBench[T, Coll <: ParIterable[T]] extends ParIterableBen
   }
 
   class Map(val size: Int, val parallelism: Int, val runWhat: String)
-  extends IterableBench with StandardParIterableBench[T, Coll] {
+  extends IterableBench {
     def comparisonMap = collection.Map()
     def runseq = this.seqcoll.map(operators.mapper)
     def runpar = this.parcoll.map(operators.mapper)
@@ -128,33 +120,108 @@ trait StandardParIterableBench[T, Coll <: ParIterable[T]] extends ParIterableBen
 
 
 
-/**
- * Benchmarks for sequence views.
+/** Benchmarks for sequence views.
  */
-trait ParSeqViewBench[T, Coll <: ParSeqView[T, ParSeq[T], CollSeq], CollSeq] extends ParSeqBench[T, Coll] {
+trait ParSeqViewBenches[T, Coll <: ParSeqView[T, ParSeq[T], CollSeq], CollSeq] extends ParSeqBenches[T, Coll] {
+self =>
 
-  object Reduce extends IterableBenchCompanion {
+  trait SeqViewBench extends SeqBench {
+    lazy val seqview: SeqView[T, Seq[T]] = createSeqView(size, parallelism)
+
+    def createSeqView(sz: Int, p: Int) = self.createSeqView(sz, p)
+  }
+
+  def createSeqView(sz: Int, p: Int): SeqView[T, Seq[T]]
+
+  object Iteration extends SeqBenchCompanion {
+    override def defaultSize = 250000
+    def benchName = "iter"
+    def apply(sz: Int, p: Int, w: String) = new Iteration(sz, p, w)
+  }
+
+  class Iteration(val size: Int, val parallelism: Int, val runWhat: String)
+  extends SeqBench with SeqViewBench {
+    def comparisonMap = collection.Map("seqview" -> runseqview _)
+    def runseq = this.seqcoll.foreach(operators.eachFun)
+    def runpar = this.parcoll.foreach(operators.eachFun)
+    def runseqview = {
+      this.seqview.foreach(operators.eachFun)
+    }
+    def companion = Iteration
+  }
+
+  object IterationS extends SeqBenchCompanion {
+    override def defaultSize = 250000
+    def benchName = "iter-s"
+    def apply(sz: Int, p: Int, w: String) = new IterationS(sz, p, w)
+  }
+
+  class IterationS(val size: Int, val parallelism: Int, val runWhat: String)
+  extends SeqBench with SeqViewBench {
+    def comparisonMap = collection.Map("seqview" -> runseqview _)
+    def runseq = this.seqcoll.slice(0, size / 2).foreach(operators.eachFun)
+    def runpar = this.parcoll.slice(0, size / 2).foreach(operators.eachFun)
+    def runseqview = this.seqview.slice(0, size / 2).foreach(operators.eachFun)
+    def companion = IterationS
+  }
+
+  object IterationM extends SeqBenchCompanion {
+    override def defaultSize = 100000
+    def benchName = "iter-m"
+    def apply(sz: Int, p: Int, w: String) = new IterationM(sz, p, w)
+  }
+
+  class IterationM(val size: Int, val parallelism: Int, val runWhat: String)
+  extends SeqBench with SeqViewBench {
+    def comparisonMap = collection.Map("seqview" -> runseqview _)
+    def runseq = this.seqcoll.map(operators.mapper).foreach(operators.eachFun)
+    def runpar = this.parcoll.map(operators.mapper).foreach(operators.eachFun)
+    def runseqview = this.seqview.map(operators.mapper).foreach(operators.eachFun)
+    def companion = IterationM
+  }
+
+  object IterationA extends SeqBenchCompanion {
+    override def defaultSize = 50000
+    def benchName = "iter-a"
+    def apply(sz: Int, p: Int, w: String) = new IterationA(sz, p, w)
+  }
+
+  class IterationA(val size: Int, val parallelism: Int, val runWhat: String)
+  extends SeqBench with SeqViewBench {
+    val appended = operators.sequence(size)
+    val sqappended = appended.toSeq
+    def comparisonMap = collection.Map("seqview" -> runseqview _)
+    def runseq = {
+      val withapp = this.seqcoll.++(sqappended)
+      withapp.foreach(operators.eachFun)
+    }
+    def runpar = this.parcoll.++(appended).foreach(operators.eachFun)
+    def runseqview = this.seqview.++(appended).foreach(operators.eachFun)
+    def companion = IterationA
+  }
+
+  object Reduce extends SeqBenchCompanion {
     override def defaultSize = 50000
     def benchName = "reduce";
     def apply(sz: Int, p: Int, w: String) = new Reduce(sz, p, w)
   }
 
   class Reduce(val size: Int, val parallelism: Int, val runWhat: String)
-  extends SeqBench with ParSeqViewBench[T, Coll, CollSeq] {
+  extends SeqBench with SeqViewBench {
     def comparisonMap = collection.Map()
     def runseq = this.seqcoll.reduceLeft(operators.reducer)
     def runpar = this.parcoll.reduce(operators.reducer)
     def companion = Reduce
   }
 
-  object MediumReduce extends IterableBenchCompanion {
+  object MediumReduce extends SeqBenchCompanion {
     override def defaultSize = 50000
     def benchName = "reduce-medium";
     def apply(sz: Int, p: Int, w: String) = new MediumReduce(sz, p, w)
   }
 
   class MediumReduce(val size: Int, val parallelism: Int, val runWhat: String)
-  extends SeqBench with ParSeqViewBench[T, Coll, CollSeq] {
+  extends SeqBench with SeqViewBench {
     def comparisonMap = collection.Map()
     def runseq = this.seqcoll.reduceLeft(operators.mediumreducer)
     def runpar = this.parcoll.reduce(operators.mediumreducer)
@@ -168,7 +235,7 @@ trait ParSeqViewBench[T, Coll <: ParSeqView[T, ParSeq[T], CollSeq], CollSeq] ext
   }
 
   class ModifyThenReduce(val size: Int, val parallelism: Int, val runWhat: String)
-  extends SeqBench with ParSeqViewBench[T, Coll, CollSeq] {
+  extends SeqBench with SeqViewBench {
     val toadd = createSequential(size, parallelism)
     def comparisonMap = collection.Map()
     def runseq = {
@@ -189,7 +256,7 @@ trait ParSeqViewBench[T, Coll <: ParSeqView[T, ParSeq[T], CollSeq], CollSeq] ext
   }
 
   class ModifyThenForce(val size: Int, val parallelism: Int, val runWhat: String)
-  extends SeqBench with ParSeqViewBench[T, Coll, CollSeq] {
+  extends SeqBench with SeqViewBench {
     val toadd = createSequential(size, parallelism)
     def comparisonMap = collection.Map()
     def runseq = (seqcoll ++ toadd).drop(size).map(operators.mapper).++(toadd).take(size)
diff --git a/test/benchmarks/src/scala/collection/parallel/benchmarks/hashtries/ParallelHashTries.scala b/test/benchmarks/src/scala/collection/parallel/benchmarks/hashtries/ParallelHashTries.scala
index c68bce93b2..11726b05bb 100644
--- a/test/benchmarks/src/scala/collection/parallel/benchmarks/hashtries/ParallelHashTries.scala
+++ b/test/benchmarks/src/scala/collection/parallel/benchmarks/hashtries/ParallelHashTries.scala
@@ -3,8 +3,7 @@ package scala.collection.parallel.benchmarks.hashtries
 
 
 
-import scala.collection.parallel.benchmarks.generic.StandardParIterableBench
-import scala.collection.parallel.benchmarks.generic.NotBenchmark
+import scala.collection.parallel.benchmarks.generic.StandardParIterableBenches
 import scala.collection.parallel.benchmarks.generic.Dummy
 import scala.collection.parallel.benchmarks.generic.Operators
 import scala.collection.parallel.immutable.ParHashTrie
@@ -13,7 +12,7 @@ import scala.collection.parallel.immutable.ParHashTrie
 
 
 
-trait ParHashTrieBenches[K, V] extends StandardParIterableBench[(K, V), ParHashTrie[K, V]] {
+trait ParHashTrieBenches[K, V] extends StandardParIterableBenches[(K, V), ParHashTrie[K, V]] {
 
   def nameOfCollection = "ParHashTrie"
   def comparisonMap = collection.Map()
@@ -27,7 +26,7 @@ trait ParHashTrieBenches[K, V] extends StandardParIterableBench[(K, V), ParHashT
   }
 
   class Map2(val size: Int, val parallelism: Int, val runWhat: String)
-  extends IterableBench with StandardParIterableBench[(K, V), ParHashTrie[K, V]] {
+  extends IterableBench {
     var result: Int = 0
     def comparisonMap = collection.Map("jhashtable" -> runjhashtable _, "hashtable" -> runhashtable _)
     def runseq = {
@@ -78,7 +77,7 @@ trait ParHashTrieBenches[K, V] extends StandardParIterableBench[(K, V), ParHashT
   }
 
   class Reduce2(val size: Int, val parallelism: Int, val runWhat: String)
-  extends IterableBench with StandardParIterableBench[(K, V), ParHashTrie[K, V]] {
+  extends IterableBench {
     private var ht: collection.mutable.HashMap[K, V] = _
     def comparisonMap = collection.Map("hashtable" -> runhashtable _)
     def runseq = this.seqcoll.reduceLeft(operators.reducer)
@@ -99,7 +98,7 @@ trait ParHashTrieBenches[K, V] extends StandardParIterableBench[(K, V), ParHashT
 
 
 
-object RefParHashTrieBenches extends ParHashTrieBenches[Dummy, Dummy] with NotBenchmark {
+object RefParHashTrieBenches extends ParHashTrieBenches[Dummy, Dummy] {
 
   type DPair = (Dummy, Dummy)
 
@@ -149,6 +148,9 @@ object RefParHashTrieBenches extends ParHashTrieBenches[Dummy, Dummy] with NotBe
       (a, p._2)
     }
     val taker = (p: DPair) => true
+    val eachFun: DPair => Unit = { dp =>
+      dp._1.dummy
+    }
   }
 
   def createSequential(sz: Int, p: Int) = {
diff --git a/test/benchmarks/src/scala/collection/parallel/benchmarks/parallel_range/RangeBenches.scala b/test/benchmarks/src/scala/collection/parallel/benchmarks/parallel_range/RangeBenches.scala
index 8a01d668fb..14a6259a38 100644
--- a/test/benchmarks/src/scala/collection/parallel/benchmarks/parallel_range/RangeBenches.scala
+++ b/test/benchmarks/src/scala/collection/parallel/benchmarks/parallel_range/RangeBenches.scala
@@ -6,13 +6,13 @@ package scala.collection.parallel.benchmarks.parallel_range
 
 import scala.collection.parallel.benchmarks.generic._
 import scala.collection.parallel.immutable.ParRange
+import scala.collection.parallel.benchmarks.generic.StandardParIterableBenches
 
 
 
 
 
-
-object RangeBenches extends StandardParIterableBench[Int, ParRange] with NotBenchmark {
+object RangeBenches extends StandardParIterableBenches[Int, ParRange] {
 
   def nameOfCollection = "ParRange"
   def operators = new IntOperators {}
@@ -33,7 +33,7 @@ object RangeBenches extends StandardParIterableBench[Int, ParRange] with NotBenc
   }
 
   class MapLight(val size: Int, val parallelism: Int, val runWhat: String)
-  extends IterableBench with StandardParIterableBench[Int, ParRange] {
+  extends IterableBench {
     def calc(n: Int) = n % 2 + 1
 
     def comparisonMap = collection.Map()
@@ -49,7 +49,7 @@ object RangeBenches extends StandardParIterableBench[Int, ParRange] with NotBenc
   }
 
   class MapMedium(val size: Int, val parallelism: Int, val runWhat: String)
-  extends IterableBench with StandardParIterableBench[Int, ParRange] {
+  extends IterableBench {
     def calc(n: Int) = {
       var i = 0
       var sum = n
@@ -73,7 +73,7 @@ object RangeBenches extends StandardParIterableBench[Int, ParRange] with NotBenc
   }
 
   class ForeachModify(val size: Int, val parallelism: Int, val runWhat: String)
-  extends IterableBench with StandardParIterableBench[Int, ParRange] {
+  extends IterableBench {
     val array = new Array[Int](size)
     def modify(n: Int) = array(n) += 1
 
@@ -90,7 +90,7 @@ object RangeBenches extends StandardParIterableBench[Int, ParRange] with NotBenc
   }
 
   class ForeachModifyMedium(val size: Int, val parallelism: Int, val runWhat: String)
-  extends IterableBench with StandardParIterableBench[Int, ParRange] {
+  extends IterableBench {
     val array = new Array[Int](size)
     def modify(n: Int) = array(n) = {
       var i = 0
@@ -115,7 +115,7 @@ object RangeBenches extends StandardParIterableBench[Int, ParRange] with NotBenc
   }
 
   class ForeachModifyHeavy(val size: Int, val parallelism: Int, val runWhat: String)
-  extends IterableBench with StandardParIterableBench[Int, ParRange] {
+  extends IterableBench {
     val array = new Array[Int](size)
     def modify(n: Int) = array(n) = collatz(10000 + array(n))
 
@@ -133,7 +133,7 @@ object RangeBenches extends StandardParIterableBench[Int, ParRange] with NotBenc
   }
 
   class ForeachAdd(val size: Int, val parallelism: Int, val runWhat: String)
-  extends IterableBench with StandardParIterableBench[Int, ParRange] {
+  extends IterableBench {
     val cmap = new java.util.concurrent.ConcurrentHashMap[Int, Int]
     val hmap = new java.util.HashMap[Int, Int]
 
@@ -157,7 +157,7 @@ object RangeBenches extends StandardParIterableBench[Int, ParRange] with NotBenc
   }
 
   class ForeachAddCollatz(val size: Int, val parallelism: Int, val runWhat: String)
-  extends IterableBench with StandardParIterableBench[Int, ParRange] {
+  extends IterableBench {
     val cmap = new java.util.concurrent.ConcurrentHashMap[Int, Int]
     val hmap = new java.util.HashMap[Int, Int]
 
diff --git a/test/benchmarks/src/scala/collection/parallel/benchmarks/parallel_view/SeqViewBenches.scala b/test/benchmarks/src/scala/collection/parallel/benchmarks/parallel_view/SeqViewBenches.scala
index e84b25971b..eed62fc5c1 100644
--- a/test/benchmarks/src/scala/collection/parallel/benchmarks/parallel_view/SeqViewBenches.scala
+++ b/test/benchmarks/src/scala/collection/parallel/benchmarks/parallel_view/SeqViewBenches.scala
@@ -16,7 +16,7 @@ import scala.collection.SeqView
 
 
 trait DummyViewBenches
-extends ParSeqViewBench[Dummy, ParSeqView[Dummy, ParSeq[Dummy], Seq[Dummy]], Seq[Dummy]] {
+extends ParSeqViewBenches[Dummy, ParSeqView[Dummy, ParSeq[Dummy], Seq[Dummy]], Seq[Dummy]] {
   def nameOfCollection = "ParView"
   def operators = DummyOperators
   def comparisonMap = collection.Map()
@@ -34,10 +34,11 @@ extends ParSeqViewBench[Dummy, ParSeqView[Dummy, ParSeq[Dummy], Seq[Dummy]], Seq
     v.environment = forkJoinPool
     v
   }
+  def createSeqView(sz: Int, p: Int) = createSequential(sz, p).view
 }
 
 
-object DummyViewBenchList extends DummyViewBenches with NotBenchmark
+object DummyViewBenchList extends DummyViewBenches