diff options
3 files changed, 13 insertions, 3 deletions
diff --git a/src/library/scala/collection/parallel/ParIterableLike.scala b/src/library/scala/collection/parallel/ParIterableLike.scala index 0769181150..ff37c89517 100644 --- a/src/library/scala/collection/parallel/ParIterableLike.scala +++ b/src/library/scala/collection/parallel/ParIterableLike.scala @@ -941,7 +941,14 @@ extends IterableLike[T, Repr] result = true } else pit.scanToArray(z, op, array, from) def newSubtask(p: ParIterator) = unsupported - override def shouldSplitFurther = len > size / 2 + override lazy val shouldSplitFurther = { + // we want less work stealings while prefix scanning + // and keep processors busier with merging + // (work stealing while prefix scanning means more work later) + val processors = parallelismLevel min availableProcessors + len > ((size / tweak(processors)) max 1) + } + private def tweak(p: Int) = if (p < 4) 2 else p / 2 override def split = { val pits = pit.split for ((p, untilp) <- pits zip pits.scanLeft(0)(_ + _.remaining); if untilp < len) yield { diff --git a/src/library/scala/collection/parallel/package.scala b/src/library/scala/collection/parallel/package.scala index c0379b819e..3d8e7208ae 100644 --- a/src/library/scala/collection/parallel/package.scala +++ b/src/library/scala/collection/parallel/package.scala @@ -12,6 +12,7 @@ import scala.collection.generic.CanCombineFrom package object parallel { val MIN_FOR_COPY = -1 // TODO: set to 5000 val CHECK_RATE = 512 + val SQRT2 = math.sqrt(2) /** Computes threshold from the size of the collection and the parallelism level. */ @@ -21,6 +22,8 @@ package object parallel { else sz } + val availableProcessors = java.lang.Runtime.getRuntime.availableProcessors + def unsupported(msg: String) = throw new UnsupportedOperationException(msg) def unsupported = throw new UnsupportedOperationException @@ -30,7 +33,7 @@ package object parallel { * * @tparam T type of the elements in the array, which is a subtype of AnyRef * @param array the array to be parallelized - * @return a `Parallelizable` object with a `par` method + * @return a `Parallelizable` object with a `par` method= */ implicit def array2ParArray[T <: AnyRef](array: Array[T]) = new Parallelizable[mutable.ParArray[T]] { def par = mutable.ParArray.handoff[T](array) diff --git a/test/benchmarks/src/scala/collection/parallel/benchmarks/parallel_array/ScanLight.scala b/test/benchmarks/src/scala/collection/parallel/benchmarks/parallel_array/ScanLight.scala index 0aa9a71cb8..ad2b534fd7 100644 --- a/test/benchmarks/src/scala/collection/parallel/benchmarks/parallel_array/ScanLight.scala +++ b/test/benchmarks/src/scala/collection/parallel/benchmarks/parallel_array/ScanLight.scala @@ -20,7 +20,7 @@ object ScanLight extends Companion { class ScanLight(sz: Int, p: Int, what: String) extends Resettable[Cont](sz, p, what, new Cont(_), new Array[Any](_), classOf[Cont]) { - def companion = AggregateLight + def companion = ScanLight override def repetitionsPerRun = 10 override val runs = 10 |