Adding special take and drop for numeric ranges...

Adding special take and drop for numeric ranges, and a test. Parallel
numeric ranges are added, but currently disabled.

Review by extempore.

4 files changed, 161 insertions, 2 deletions

diff --git a/src/library/scala/collection/immutable/NumericRange.scala b/src/library/scala/collection/immutable/NumericRange.scala
index fb5c7221dc..a548b62533 100644
--- a/src/library/scala/collection/immutable/NumericRange.scala
+++ b/src/library/scala/collection/immutable/NumericRange.scala
@@ -42,7 +42,7 @@ import generic._
 abstract class NumericRange[T]
   (val start: T, val end: T, val step: T, val isInclusive: Boolean)
   (implicit num: Integral[T])
-extends IndexedSeq[T] with Serializable {
+extends IndexedSeq[T] with Parallelizable[parallel.immutable.ParNumericRange[T]] with Serializable {
   /** Note that NumericRange must be invariant so that constructs
    *  such as "1L to 10 by 5" do not infer the range type as AnyVal.
    */
@@ -57,6 +57,8 @@ extends IndexedSeq[T] with Serializable {
     if (length == 0) Nil.last
     else locationAfterN(length - 1)
 
+  def par = new parallel.immutable.ParNumericRange(this)
+
   /** Create a new range with the start and end values of this range and
    *  a new `step`.
    */
@@ -107,6 +109,18 @@ extends IndexedSeq[T] with Serializable {
   // based on the given value.
   private def newEmptyRange(value: T) = NumericRange(value, value, step)
 
+  final override def take(n: Int): NumericRange[T] = (
+    if (n <= 0 || length == 0) newEmptyRange(start)
+    else if (n >= length) this
+    else new NumericRange.Inclusive(start, locationAfterN(n - 1), step)
+  )
+
+  final override def drop(n: Int): NumericRange[T] = (
+    if (n <= 0 || length == 0) this
+    else if (n >= length) newEmptyRange(end)
+    else copy(locationAfterN(n), end, step)
+  )
+
   def apply(idx: Int): T = {
     if (idx < 0 || idx >= length) throw new IndexOutOfBoundsException(idx.toString)
     else locationAfterN(idx)
diff --git a/src/library/scala/collection/parallel/ParIterableLike.scala b/src/library/scala/collection/parallel/ParIterableLike.scala
index b51619e657..a92c4f15fb 100644
--- a/src/library/scala/collection/parallel/ParIterableLike.scala
+++ b/src/library/scala/collection/parallel/ParIterableLike.scala
@@ -462,7 +462,7 @@ self =>
   }
 
   override def minBy[S](f: T => S)(implicit cmp: Ordering[S]): T = {
-    if (isEmpty) throw new UnsupportedOperationException("empty.maxBy")
+    if (isEmpty) throw new UnsupportedOperationException("empty.minBy")
 
     reduce((x, y) => if (cmp.lteq(f(x), f(y))) x else y)
   }
diff --git a/src/library/scala/collection/parallel/immutable/ParNumericRange.scala.disabled b/src/library/scala/collection/parallel/immutable/ParNumericRange.scala.disabled
new file mode 100644
index 0000000000..a32d7bb086
--- /dev/null
+++ b/src/library/scala/collection/parallel/immutable/ParNumericRange.scala.disabled
@@ -0,0 +1,132 @@
+/*                     __                                               *\
+**     ________ ___   / /  ___     Scala API                            **
+**    / __/ __// _ | / /  / _ |    (c) 2003-2011, LAMP/EPFL             **
+**  __\ \/ /__/ __ |/ /__/ __ |    http://scala-lang.org/               **
+** /____/\___/_/ |_/____/_/ | |                                         **
+**                          |/                                          **
+\*                                                                      */
+
+
+package scala.collection.parallel.immutable
+
+
+
+import scala.collection.immutable.NumericRange
+import scala.collection.parallel.Combiner
+import scala.collection.generic.CanCombineFrom
+import scala.collection.parallel.ParIterableIterator
+
+
+
+/** Parallel ranges for numeric types.
+ *  
+ *  $paralleliterableinfo
+ *  
+ *  $sideeffects
+ *  
+ *  @param range    the sequential range this parallel range was obtained from
+ *  
+ *  @author Aleksandar Prokopec
+ *  @since 2.9
+ *  
+ *  @define Coll immutable.ParRange
+ *  @define coll immutable parallel range
+ */
+@SerialVersionUID(1L)
+class ParNumericRange[T](val range: NumericRange[T])(implicit num: Integral[T])
+extends ParSeq[T]
+   with Serializable
+{
+self =>
+  
+  def seq = range
+  
+  @inline final def length = range.length
+  
+  @inline final def apply(idx: Int) = range.apply(idx);
+  
+  def parallelIterator = new ParNumericRangeIterator with SCPI
+  
+  type SCPI = SignalContextPassingIterator[ParNumericRangeIterator]
+  
+  override def toParSeq = this
+  
+  override def toParSet[U >: T] = toParCollection[U, ParSet[U]](() => HashSetCombiner[U])
+  
+  class ParNumericRangeIterator(range: NumericRange[T] = self.range, num: Integral[T] = self.num)
+  extends ParIterator {
+  me: SignalContextPassingIterator[ParNumericRangeIterator] =>
+    override def toString = "ParNumericRangeIterator(over: " + range + ")"
+    private var ind = 0
+    private val len = range.length
+    
+    final def remaining = len - ind
+    
+    final def hasNext = ind < len
+    
+    final def next = if (hasNext) {
+      val r = range.apply(ind)
+      ind += 1
+      r
+    } else Iterator.empty.next
+    
+    private def rangeleft: NumericRange[T] = range.drop(ind)
+    
+    def dup = new ParNumericRangeIterator(rangeleft) with SCPI
+    
+    def split = {
+      val rleft = rangeleft
+      val elemleft = rleft.length
+      if (elemleft < 2) Seq(new ParNumericRangeIterator(rleft) with SCPI)
+      else Seq(
+        new ParNumericRangeIterator(rleft.take(elemleft / 2)) with SCPI,
+        new ParNumericRangeIterator(rleft.drop(elemleft / 2)) with SCPI
+      )
+    }
+    
+    def psplit(sizes: Int*) = {
+      var rleft = rangeleft
+      for (sz <- sizes) yield {
+        val fronttaken = rleft.take(sz)
+        rleft = rleft.drop(sz)
+        new ParNumericRangeIterator(fronttaken) with SCPI
+      }
+    }
+    
+    /* accessors */
+    
+    override def foreach[U](f: T => U): Unit = {
+      rangeleft.foreach(f)
+      ind = len
+    }
+    
+    override def reduce[U >: T](op: (U, U) => U): U = {
+      val r = rangeleft.reduceLeft(op)
+      ind = len
+      r
+    }
+    
+    /* transformers */
+    
+    override def map2combiner[S, That](f: T => S, cb: Combiner[S, That]): Combiner[S, That] = {
+      while (hasNext) {
+        cb += f(next)
+      }
+      cb
+    }
+  }
+  
+}
+
+
+object ParNumericRange {
+  def apply[T](start: T, end: T, step: T, inclusive: Boolean)(implicit num: Integral[T]) = new ParNumericRange[T](
+    if (inclusive) NumericRange.inclusive(start, end, step)(num)
+    else NumericRange.apply(start, end, step)(num)
+  )
+}
+
+
+
+
+
diff --git a/test/files/run/numeric-range.scala b/test/files/run/numeric-range.scala
new file mode 100644
index 0000000000..4645db6ef0
--- /dev/null
+++ b/test/files/run/numeric-range.scala
@@ -0,0 +1,13 @@
+
+
+
+
+object Test {
+  def main(args: Array[String]) {
+    val r = 'a' to 'z'
+    for (i <- -2 to (r.length + 2)) {
+      assert(r.take(i) == r.toList.take(i), (i, r.take(i)))
+      assert(r.drop(i) == r.toList.drop(i), (i, r.drop(i)))
+    }
+  }
+}