diff options
14 files changed, 113 insertions, 182 deletions
diff --git a/src/library/scala/collection/parallel/ParIterableLike.scala b/src/library/scala/collection/parallel/ParIterableLike.scala index 75f4552076..b24497371d 100644 --- a/src/library/scala/collection/parallel/ParIterableLike.scala +++ b/src/library/scala/collection/parallel/ParIterableLike.scala @@ -171,42 +171,6 @@ self: ParIterableLike[T, Repr, Sequential] => def seq: Sequential def repr: Repr = this.asInstanceOf[Repr] - - /** Parallel iterators are split iterators that have additional accessor and - * transformer methods defined in terms of methods `next` and `hasNext`. - * When creating a new parallel collection, one might want to override these - * new methods to make them more efficient. - * - * Parallel iterators are augmented with signalling capabilities. This means - * that a signalling object can be assigned to them as needed. - * - * The self-type ensures that signal context passing behaviour gets mixed in - * a concrete object instance. - */ - trait ParIterator extends IterableSplitter[T] { - me: SignalContextPassingIterator[ParIterator] => - var signalDelegate: Signalling = IdleSignalling - def repr = self.repr - def split: Seq[IterableSplitter[T]] - } - - /** A stackable modification that ensures signal contexts get passed along the iterators. - * A self-type requirement in `ParIterator` ensures that this trait gets mixed into - * concrete iterators. - */ - trait SignalContextPassingIterator[+IterRepr <: ParIterator] extends ParIterator { - // Note: This functionality must be factored out to this inner trait to avoid boilerplate. - // Also, one could omit the cast below. However, this leads to return type inconsistencies, - // due to inability to override the return type of _abstract overrides_. - // Be aware that this stackable modification has to be subclassed, so it shouldn't be rigid - // on the type of iterators it splits. - // The alternative is some boilerplate - better to tradeoff some type safety to avoid it here. - abstract override def split: Seq[IterRepr] = { - val pits = super.split - pits foreach { _.signalDelegate = signalDelegate } - pits.asInstanceOf[Seq[IterRepr]] - } - } def hasDefiniteSize = true @@ -870,7 +834,7 @@ self: ParIterableLike[T, Repr, Sequential] => protected[this] val pit: IterableSplitter[T] protected[this] def newSubtask(p: IterableSplitter[T]): Accessor[R, Tp] def shouldSplitFurther = pit.remaining > threshold(size, parallelismLevel) - def split = pit.split.map(newSubtask(_)) // default split procedure + def split = pit.splitWithSignalling.map(newSubtask(_)) // default split procedure private[parallel] override def signalAbort = pit.abort override def toString = this.getClass.getSimpleName + "(" + pit.toString + ")(" + result + ")(supername: " + super.toString + ")" } @@ -1142,7 +1106,7 @@ self: ParIterableLike[T, Repr, Sequential] => } protected[this] def newSubtask(p: IterableSplitter[T]) = throw new UnsupportedOperationException override def split = { - val pits = pit.split + val pits = pit.splitWithSignalling val sizes = pits.scanLeft(0)(_ + _.remaining) for ((p, untilp) <- pits zip sizes; if untilp <= n) yield { if (untilp + p.remaining < n) new Take(p.remaining, cbf, p) @@ -1160,7 +1124,7 @@ self: ParIterableLike[T, Repr, Sequential] => def leaf(prev: Option[Combiner[U, This]]) = result = pit.drop2combiner(n, reuse(prev, cbf())) protected[this] def newSubtask(p: IterableSplitter[T]) = throw new UnsupportedOperationException override def split = { - val pits = pit.split + val pits = pit.splitWithSignalling val sizes = pits.scanLeft(0)(_ + _.remaining) for ((p, withp) <- pits zip sizes.tail; if withp >= n) yield { if (withp - p.remaining > n) new Drop(0, cbf, p) @@ -1178,7 +1142,7 @@ self: ParIterableLike[T, Repr, Sequential] => def leaf(prev: Option[Combiner[U, This]]) = result = pit.slice2combiner(from, until, reuse(prev, cbf())) protected[this] def newSubtask(p: IterableSplitter[T]) = throw new UnsupportedOperationException override def split = { - val pits = pit.split + val pits = pit.splitWithSignalling val sizes = pits.scanLeft(0)(_ + _.remaining) for ((p, untilp) <- pits zip sizes; if untilp + p.remaining >= from || untilp <= until) yield { val f = (from max untilp) - untilp @@ -1197,7 +1161,7 @@ self: ParIterableLike[T, Repr, Sequential] => def leaf(prev: Option[(Combiner[U, This], Combiner[U, This])]) = result = pit.splitAt2combiners(at, reuse(prev.map(_._1), cbfBefore()), reuse(prev.map(_._2), cbfAfter())) protected[this] def newSubtask(p: IterableSplitter[T]) = throw new UnsupportedOperationException override def split = { - val pits = pit.split + val pits = pit.splitWithSignalling val sizes = pits.scanLeft(0)(_ + _.remaining) for ((p, untilp) <- pits zip sizes) yield new SplitAt((at max untilp min (untilp + p.remaining)) - untilp, cbfBefore, cbfAfter, p) } @@ -1215,7 +1179,7 @@ self: ParIterableLike[T, Repr, Sequential] => } else result = (reuse(prev.map(_._1), cbf()), false) protected[this] def newSubtask(p: IterableSplitter[T]) = throw new UnsupportedOperationException override def split = { - val pits = pit.split + val pits = pit.splitWithSignalling for ((p, untilp) <- pits zip pits.scanLeft(0)(_ + _.remaining)) yield new TakeWhile(pos + untilp, pred, cbf, p) } override def merge(that: TakeWhile[U, This]) = if (result._2) { @@ -1240,7 +1204,7 @@ self: ParIterableLike[T, Repr, Sequential] => } protected[this] def newSubtask(p: IterableSplitter[T]) = throw new UnsupportedOperationException override def split = { - val pits = pit.split + val pits = pit.splitWithSignalling for ((p, untilp) <- pits zip pits.scanLeft(0)(_ + _.remaining)) yield new Span(pos + untilp, pred, cbfBefore, cbfAfter, p) } override def merge(that: Span[U, This]) = result = if (result._2.size == 0) { @@ -1257,9 +1221,9 @@ self: ParIterableLike[T, Repr, Sequential] => def leaf(prev: Option[Result]) = result = pit.zip2combiner[U, S, That](othpit, pbf()) protected[this] def newSubtask(p: IterableSplitter[T]) = unsupported override def split = { - val pits = pit.split + val pits = pit.splitWithSignalling val sizes = pits.map(_.remaining) - val opits = othpit.psplit(sizes: _*) + val opits = othpit.psplitWithSignalling(sizes: _*) (pits zip opits) map { p => new Zip(pbf, p._1, p._2) } } override def merge(that: Zip[U, S, That]) = result = result combine that.result @@ -1273,12 +1237,12 @@ self: ParIterableLike[T, Repr, Sequential] => def leaf(prev: Option[Result]) = result = pit.zipAll2combiner[U, S, That](othpit, thiselem, thatelem, pbf()) protected[this] def newSubtask(p: IterableSplitter[T]) = unsupported override def split = if (pit.remaining <= len) { - val pits = pit.split + val pits = pit.splitWithSignalling val sizes = pits.map(_.remaining) - val opits = othpit.psplit(sizes: _*) + val opits = othpit.psplitWithSignalling(sizes: _*) ((pits zip opits) zip sizes) map { t => new ZipAll(t._2, thiselem, thatelem, pbf, t._1._1, t._1._2) } } else { - val opits = othpit.psplit(pit.remaining) + val opits = othpit.psplitWithSignalling(pit.remaining) val diff = len - pit.remaining Seq( new ZipAll(pit.remaining, thiselem, thatelem, pbf, pit, opits(0)), // nothing wrong will happen with the cast below - elem T is never accessed @@ -1295,7 +1259,7 @@ self: ParIterableLike[T, Repr, Sequential] => def leaf(prev: Option[Unit]) = pit.copyToArray(array, from, len) protected[this] def newSubtask(p: IterableSplitter[T]) = unsupported override def split = { - val pits = pit.split + val pits = pit.splitWithSignalling for ((p, untilp) <- pits zip pits.scanLeft(0)(_ + _.remaining); if untilp < len) yield { val plen = p.remaining min (len - untilp) new CopyToArray[U, This](from + untilp, plen, array, p) @@ -1352,7 +1316,7 @@ self: ParIterableLike[T, Repr, Sequential] => } else trees(from) protected[this] def newSubtask(pit: IterableSplitter[T]) = unsupported override def split = { - val pits = pit.split + val pits = pit.splitWithSignalling for ((p, untilp) <- pits zip pits.scanLeft(from)(_ + _.remaining)) yield { new CreateScanTree(untilp, p.remaining, z, op, p) } diff --git a/src/library/scala/collection/parallel/ParMapLike.scala b/src/library/scala/collection/parallel/ParMapLike.scala index beb50a41e1..afd1f30903 100644 --- a/src/library/scala/collection/parallel/ParMapLike.scala +++ b/src/library/scala/collection/parallel/ParMapLike.scala @@ -66,7 +66,6 @@ self => new IterableSplitter[K] { i => val iter = s - var signalDelegate: Signalling = IdleSignalling def hasNext = iter.hasNext def next() = iter.next._1 def split = { @@ -84,7 +83,6 @@ self => new IterableSplitter[V] { i => val iter = s - var signalDelegate: Signalling = IdleSignalling def hasNext = iter.hasNext def next() = iter.next._2 def split = { diff --git a/src/library/scala/collection/parallel/ParSeqLike.scala b/src/library/scala/collection/parallel/ParSeqLike.scala index 22c587b498..6a5ee5c69b 100644 --- a/src/library/scala/collection/parallel/ParSeqLike.scala +++ b/src/library/scala/collection/parallel/ParSeqLike.scala @@ -48,35 +48,6 @@ self => type SuperParIterator = IterableSplitter[T] - /** An iterator that can be split into arbitrary subsets of iterators. - * The self-type requirement ensures that the signal context passing behaviour gets mixed in - * the concrete iterator instance in some concrete collection. - * - * '''Note:''' In concrete collection classes, collection implementers might want to override the iterator - * `reverse2builder` method to ensure higher efficiency. - */ - trait ParIterator extends SeqSplitter[T] with super.ParIterator { - me: SignalContextPassingIterator[ParIterator] => - def split: Seq[ParIterator] - def psplit(sizes: Int*): Seq[ParIterator] - } - - /** A stackable modification that ensures signal contexts get passed along the iterators. - * A self-type requirement in `ParIterator` ensures that this trait gets mixed into - * concrete iterators. - */ - trait SignalContextPassingIterator[+IterRepr <: ParIterator] - extends ParIterator with super.SignalContextPassingIterator[IterRepr] { - // Note: See explanation in `ParallelIterableLike.this.SignalContextPassingIterator` - // to understand why we do the cast here, and have a type parameter. - // Bottomline: avoiding boilerplate and fighting against inability to override stackable modifications. - abstract override def psplit(sizes: Int*): Seq[IterRepr] = { - val pits = super.psplit(sizes: _*) - pits foreach { _.signalDelegate = signalDelegate } - pits.asInstanceOf[Seq[IterRepr]] - } - } - /** A more refined version of the iterator found in the `ParallelIterable` trait, * this iterator can be split into arbitrary subsets of iterators. * @@ -89,9 +60,7 @@ self => override def size = length /** Used to iterate elements using indices */ - protected abstract class Elements(start: Int, val end: Int) extends ParIterator with BufferedIterator[T] { - me: SignalContextPassingIterator[ParIterator] => - + protected abstract class Elements(start: Int, val end: Int) extends SeqSplitter[T] with BufferedIterator[T] { private var i = start def hasNext = i < end @@ -106,14 +75,14 @@ self => final def remaining = end - i - def dup = new Elements(i, end) with SignalContextPassingIterator[ParIterator] + def dup = new Elements(i, end) {} def split = psplit(remaining / 2, remaining - remaining / 2) def psplit(sizes: Int*) = { val incr = sizes.scanLeft(0)(_ + _) for ((from, until) <- incr.init zip incr.tail) yield { - new Elements(start + from, (start + until) min end) with SignalContextPassingIterator[ParIterator] + new Elements(start + from, (start + until) min end) {} } } @@ -138,7 +107,7 @@ self => val realfrom = if (from < 0) 0 else from val ctx = new DefaultSignalling with AtomicIndexFlag ctx.setIndexFlag(Int.MaxValue) - executeAndWaitResult(new SegmentLength(p, 0, splitter.psplit(realfrom, length - realfrom)(1) assign ctx))._1 + executeAndWaitResult(new SegmentLength(p, 0, splitter.psplitWithSignalling(realfrom, length - realfrom)(1) assign ctx))._1 } /** Finds the first element satisfying some predicate. @@ -156,7 +125,7 @@ self => val realfrom = if (from < 0) 0 else from val ctx = new DefaultSignalling with AtomicIndexFlag ctx.setIndexFlag(Int.MaxValue) - executeAndWaitResult(new IndexWhere(p, realfrom, splitter.psplit(realfrom, length - realfrom)(1) assign ctx)) + executeAndWaitResult(new IndexWhere(p, realfrom, splitter.psplitWithSignalling(realfrom, length - realfrom)(1) assign ctx)) } /** Finds the last element satisfying some predicate. @@ -174,7 +143,7 @@ self => val until = if (end >= length) length else end + 1 val ctx = new DefaultSignalling with AtomicIndexFlag ctx.setIndexFlag(Int.MinValue) - executeAndWaitResult(new LastIndexWhere(p, 0, splitter.psplit(until, length - until)(0) assign ctx)) + executeAndWaitResult(new LastIndexWhere(p, 0, splitter.psplitWithSignalling(until, length - until)(0) assign ctx)) } def reverse: Repr = { @@ -203,7 +172,7 @@ self => else if (pthat.length > length - offset) false else { val ctx = new DefaultSignalling with VolatileAbort - executeAndWaitResult(new SameElements(splitter.psplit(offset, pthat.length)(1) assign ctx, pthat.splitter)) + executeAndWaitResult(new SameElements(splitter.psplitWithSignalling(offset, pthat.length)(1) assign ctx, pthat.splitter)) } } otherwise seq.startsWith(that, offset) @@ -226,7 +195,7 @@ self => else { val ctx = new DefaultSignalling with VolatileAbort val tlen = that.length - executeAndWaitResult(new SameElements(splitter.psplit(length - tlen, tlen)(1) assign ctx, pthat.splitter)) + executeAndWaitResult(new SameElements(splitter.psplitWithSignalling(length - tlen, tlen)(1) assign ctx, pthat.splitter)) } } otherwise seq.endsWith(that) @@ -235,7 +204,7 @@ self => if (patch.isParSeq && bf.isParallel && (size - realreplaced + patch.size) > MIN_FOR_COPY) { val that = patch.asParSeq val pbf = bf.asParallel - val pits = splitter.psplit(from, replaced, length - from - realreplaced) + val pits = splitter.psplitWithSignalling(from, replaced, length - from - realreplaced) val cfactory = combinerFactory(() => pbf(repr)) val copystart = new Copy[U, That](cfactory, pits(0)) val copymiddle = wrap { @@ -253,7 +222,7 @@ self => val from = 0 max fromarg val b = bf(repr) val repl = (r min (length - from)) max 0 - val pits = splitter.psplit(from, repl, length - from - repl) + val pits = splitter.psplitWithSignalling(from, repl, length - from - repl) b ++= pits(0) b ++= patch b ++= pits(2) @@ -373,7 +342,7 @@ self => } else result = (0, false) protected[this] def newSubtask(p: SuperParIterator) = throw new UnsupportedOperationException override def split = { - val pits = pit.split + val pits = pit.splitWithSignalling for ((p, untilp) <- pits zip pits.scanLeft(0)(_ + _.remaining)) yield new SegmentLength(pred, from + untilp, p) } override def merge(that: SegmentLength) = if (result._2) result = (result._1 + that.result._1, that.result._2) @@ -392,7 +361,7 @@ self => } protected[this] def newSubtask(p: SuperParIterator) = unsupported override def split = { - val pits = pit.split + val pits = pit.splitWithSignalling for ((p, untilp) <- pits zip pits.scanLeft(from)(_ + _.remaining)) yield new IndexWhere(pred, untilp, p) } override def merge(that: IndexWhere) = result = if (result == -1) that.result else { @@ -413,7 +382,7 @@ self => } protected[this] def newSubtask(p: SuperParIterator) = unsupported override def split = { - val pits = pit.split + val pits = pit.splitWithSignalling for ((p, untilp) <- pits zip pits.scanLeft(pos)(_ + _.remaining)) yield new LastIndexWhere(pred, untilp, p) } override def merge(that: LastIndexWhere) = result = if (result == -1) that.result else { @@ -438,7 +407,7 @@ self => override def merge(that: ReverseMap[S, That]) = result = that.result combine result } - protected[this] class SameElements[U >: T](protected[this] val pit: SeqSplitter[T], val otherpit: PreciseSplitter[U]) + protected[this] class SameElements[U >: T](protected[this] val pit: SeqSplitter[T], val otherpit: SeqSplitter[U]) extends Accessor[Boolean, SameElements[U]] { @volatile var result: Boolean = true def leaf(prev: Option[Boolean]) = if (!pit.isAborted) { @@ -449,7 +418,7 @@ self => override def split = { val fp = pit.remaining / 2 val sp = pit.remaining - fp - for ((p, op) <- pit.psplit(fp, sp) zip otherpit.psplit(fp, sp)) yield new SameElements(p, op) + for ((p, op) <- pit.psplitWithSignalling(fp, sp) zip otherpit.psplitWithSignalling(fp, sp)) yield new SameElements(p, op) } override def merge(that: SameElements[U]) = result = result && that.result override def requiresStrictSplitters = true @@ -461,7 +430,7 @@ self => def leaf(prev: Option[Combiner[U, That]]) = result = pit.updated2combiner(pos, elem, pbf()) protected[this] def newSubtask(p: SuperParIterator) = unsupported override def split = { - val pits = pit.split + val pits = pit.splitWithSignalling for ((p, untilp) <- pits zip pits.scanLeft(0)(_ + _.remaining)) yield new Updated(pos - untilp, elem, pbf, p) } override def merge(that: Updated[U, That]) = result = result combine that.result @@ -476,8 +445,8 @@ self => override def split = { val fp = len / 2 val sp = len - len / 2 - val pits = pit.psplit(fp, sp) - val opits = otherpit.psplit(fp, sp) + val pits = pit.psplitWithSignalling(fp, sp) + val opits = otherpit.psplitWithSignalling(fp, sp) Seq( new Zip(fp, pbf, pits(0), opits(0)), new Zip(sp, pbf, pits(1), opits(1)) @@ -486,7 +455,7 @@ self => override def merge(that: Zip[U, S, That]) = result = result combine that.result } - protected[this] class Corresponds[S](corr: (T, S) => Boolean, protected[this] val pit: SeqSplitter[T], val otherpit: PreciseSplitter[S]) + protected[this] class Corresponds[S](corr: (T, S) => Boolean, protected[this] val pit: SeqSplitter[T], val otherpit: SeqSplitter[S]) extends Accessor[Boolean, Corresponds[S]] { @volatile var result: Boolean = true def leaf(prev: Option[Boolean]) = if (!pit.isAborted) { @@ -497,7 +466,7 @@ self => override def split = { val fp = pit.remaining / 2 val sp = pit.remaining - fp - for ((p, op) <- pit.psplit(fp, sp) zip otherpit.psplit(fp, sp)) yield new Corresponds(corr, p, op) + for ((p, op) <- pit.psplitWithSignalling(fp, sp) zip otherpit.psplitWithSignalling(fp, sp)) yield new Corresponds(corr, p, op) } override def merge(that: Corresponds[S]) = result = result && that.result override def requiresStrictSplitters = true diff --git a/src/library/scala/collection/parallel/RemainsIterator.scala b/src/library/scala/collection/parallel/RemainsIterator.scala index e04e0e9c72..e8b332da89 100644 --- a/src/library/scala/collection/parallel/RemainsIterator.scala +++ b/src/library/scala/collection/parallel/RemainsIterator.scala @@ -14,6 +14,7 @@ package scala.collection.parallel import scala.collection.Parallel import scala.collection.generic.Signalling import scala.collection.generic.DelegatedSignalling +import scala.collection.generic.IdleSignalling import scala.collection.generic.CanCombineFrom import scala.collection.mutable.Builder import scala.collection.Iterator.empty @@ -380,12 +381,20 @@ extends AugmentedIterableIterator[T] with DelegatedSignalling { self => - + + var signalDelegate: Signalling = IdleSignalling + /** Creates a copy of this iterator. */ def dup: IterableSplitter[T] def split: Seq[IterableSplitter[T]] - + + def splitWithSignalling: Seq[IterableSplitter[T]] = { + val pits = split + pits foreach { _.signalDelegate = signalDelegate } + pits + } + /** The number of elements this iterator has yet to traverse. This method * doesn't change the state of the iterator. * @@ -421,7 +430,6 @@ self => /* iterator transformers */ class Taken(taken: Int) extends IterableSplitter[T] { - var signalDelegate = self.signalDelegate var remaining = taken min self.remaining def hasNext = remaining > 0 def next = { remaining -= 1; self.next } @@ -450,7 +458,7 @@ self => override def slice(from1: Int, until1: Int): IterableSplitter[T] = newSliceInternal(newTaken(until1), from1) class Mapped[S](f: T => S) extends IterableSplitter[S] { - var signalDelegate = self.signalDelegate + signalDelegate = self.signalDelegate def hasNext = self.hasNext def next = f(self.next) def remaining = self.remaining @@ -461,7 +469,7 @@ self => override def map[S](f: T => S) = new Mapped(f) class Appended[U >: T, PI <: IterableSplitter[U]](protected val that: PI) extends IterableSplitter[U] { - var signalDelegate = self.signalDelegate + signalDelegate = self.signalDelegate protected var curr: IterableSplitter[U] = self def hasNext = if (curr.hasNext) true else if (curr eq self) { curr = that @@ -480,7 +488,7 @@ self => def appendParIterable[U >: T, PI <: IterableSplitter[U]](that: PI) = new Appended[U, PI](that) class Zipped[S](protected val that: SeqSplitter[S]) extends IterableSplitter[(T, S)] { - var signalDelegate = self.signalDelegate + signalDelegate = self.signalDelegate def hasNext = self.hasNext && that.hasNext def next = (self.next, that.next) def remaining = self.remaining min that.remaining @@ -497,7 +505,7 @@ self => class ZippedAll[U >: T, S](protected val that: SeqSplitter[S], protected val thiselem: U, protected val thatelem: S) extends IterableSplitter[(U, S)] { - var signalDelegate = self.signalDelegate + signalDelegate = self.signalDelegate def hasNext = self.hasNext || that.hasNext def next = if (self.hasNext) { if (that.hasNext) (self.next, that.next) @@ -534,6 +542,18 @@ self => def split: Seq[SeqSplitter[T]] def psplit(sizes: Int*): Seq[SeqSplitter[T]] + override def splitWithSignalling: Seq[SeqSplitter[T]] = { + val pits = split + pits foreach { _.signalDelegate = signalDelegate } + pits + } + + def psplitWithSignalling(sizes: Int*): Seq[SeqSplitter[T]] = { + val pits = psplit(sizes: _*) + pits foreach { _.signalDelegate = signalDelegate } + pits + } + /** The number of elements this iterator has yet to traverse. This method * doesn't change the state of the iterator. Unlike the version of this method in the supertrait, * method `remaining` in `ParSeqLike.this.ParIterator` must return an exact number @@ -626,13 +646,13 @@ self => def reverse: SeqSplitter[T] = { val pa = mutable.ParArray.fromTraversables(self).reverse - new pa.ParArrayIterator with pa.SCPI { + new pa.ParArrayIterator { override def reverse = self } } class Patched[U >: T](from: Int, patch: SeqSplitter[U], replaced: Int) extends SeqSplitter[U] { - var signalDelegate = self.signalDelegate + signalDelegate = self.signalDelegate private[this] val trio = { val pits = self.psplit(from, replaced, self.remaining - from - replaced) (pits(0).appendParSeq[U, SeqSplitter[U]](patch)) appendParSeq pits(2) diff --git a/src/library/scala/collection/parallel/immutable/ParHashMap.scala b/src/library/scala/collection/parallel/immutable/ParHashMap.scala index e785932933..7adf51cffb 100644 --- a/src/library/scala/collection/parallel/immutable/ParHashMap.scala +++ b/src/library/scala/collection/parallel/immutable/ParHashMap.scala @@ -52,7 +52,7 @@ self => protected[this] override def newCombiner = HashMapCombiner[K, V] - def splitter: IterableSplitter[(K, V)] = new ParHashMapIterator(trie.iterator, trie.size) with SCPI + def splitter: IterableSplitter[(K, V)] = new ParHashMapIterator(trie.iterator, trie.size) override def seq = trie @@ -69,11 +69,8 @@ self => case None => newc } - type SCPI = SignalContextPassingIterator[ParHashMapIterator] - class ParHashMapIterator(var triter: Iterator[(K, V @uncheckedVariance)], val sz: Int) - extends super.ParIterator { - self: SignalContextPassingIterator[ParHashMapIterator] => + extends IterableSplitter[(K, V)] { var i = 0 def dup = triter match { case t: TrieIterator[_] => @@ -84,24 +81,24 @@ self => dupFromIterator(buff.iterator) } private def dupFromIterator(it: Iterator[(K, V @uncheckedVariance)]) = { - val phit = new ParHashMapIterator(it, sz) with SCPI + val phit = new ParHashMapIterator(it, sz) phit.i = i phit } - def split: Seq[ParIterator] = if (remaining < 2) Seq(this) else triter match { + def split: Seq[IterableSplitter[(K, V)]] = if (remaining < 2) Seq(this) else triter match { case t: TrieIterator[_] => val previousRemaining = remaining val ((fst, fstlength), snd) = t.split val sndlength = previousRemaining - fstlength Seq( - new ParHashMapIterator(fst, fstlength) with SCPI, - new ParHashMapIterator(snd, sndlength) with SCPI + new ParHashMapIterator(fst, fstlength), + new ParHashMapIterator(snd, sndlength) ) case _ => // iterator of the collision map case val buff = triter.toBuffer val (fp, sp) = buff.splitAt(buff.length / 2) - Seq(fp, sp) map { b => new ParHashMapIterator(b.iterator, b.length) with SCPI } + Seq(fp, sp) map { b => new ParHashMapIterator(b.iterator, b.length) } } def next(): (K, V) = { i += 1 diff --git a/src/library/scala/collection/parallel/immutable/ParHashSet.scala b/src/library/scala/collection/parallel/immutable/ParHashSet.scala index 8332167b90..1cf0ccd391 100644 --- a/src/library/scala/collection/parallel/immutable/ParHashSet.scala +++ b/src/library/scala/collection/parallel/immutable/ParHashSet.scala @@ -49,7 +49,7 @@ self => override def empty: ParHashSet[T] = new ParHashSet[T] - def splitter: IterableSplitter[T] = new ParHashSetIterator(trie.iterator, trie.size) with SCPI + def splitter: IterableSplitter[T] = new ParHashSetIterator(trie.iterator, trie.size) override def seq = trie @@ -66,11 +66,8 @@ self => case None => newc } - type SCPI = SignalContextPassingIterator[ParHashSetIterator] - class ParHashSetIterator(var triter: Iterator[T], val sz: Int) - extends super.ParIterator { - self: SignalContextPassingIterator[ParHashSetIterator] => + extends IterableSplitter[T] { var i = 0 def dup = triter match { case t: TrieIterator[_] => @@ -81,24 +78,24 @@ self => dupFromIterator(buff.iterator) } private def dupFromIterator(it: Iterator[T]) = { - val phit = new ParHashSetIterator(it, sz) with SCPI + val phit = new ParHashSetIterator(it, sz) phit.i = i phit } - def split: Seq[ParIterator] = if (remaining < 2) Seq(this) else triter match { + def split: Seq[IterableSplitter[T]] = if (remaining < 2) Seq(this) else triter match { case t: TrieIterator[_] => val previousRemaining = remaining val ((fst, fstlength), snd) = t.split val sndlength = previousRemaining - fstlength Seq( - new ParHashSetIterator(fst, fstlength) with SCPI, - new ParHashSetIterator(snd, sndlength) with SCPI + new ParHashSetIterator(fst, fstlength), + new ParHashSetIterator(snd, sndlength) ) case _ => // iterator of the collision map case val buff = triter.toBuffer val (fp, sp) = buff.splitAt(buff.length / 2) - Seq(fp, sp) map { b => new ParHashSetIterator(b.iterator, b.length) with SCPI } + Seq(fp, sp) map { b => new ParHashSetIterator(b.iterator, b.length) } } def next(): T = { i += 1 @@ -111,6 +108,7 @@ self => } } + /** $factoryInfo * @define Coll immutable.ParHashSet * @define coll immutable parallel hash set @@ -124,6 +122,7 @@ object ParHashSet extends ParSetFactory[ParHashSet] { def fromTrie[T](t: HashSet[T]) = new ParHashSet(t) } + private[immutable] abstract class HashSetCombiner[T] extends collection.parallel.BucketCombiner[T, ParHashSet[T], Any, HashSetCombiner[T]](HashSetCombiner.rootsize) { //self: EnvironmentPassingCombiner[T, ParHashSet[T]] => @@ -207,6 +206,7 @@ extends collection.parallel.BucketCombiner[T, ParHashSet[T], Any, HashSetCombine } } + object HashSetCombiner { def apply[T] = new HashSetCombiner[T] {} // was: with EnvironmentPassingCombiner[T, ParHashSet[T]] {} diff --git a/src/library/scala/collection/parallel/immutable/ParRange.scala b/src/library/scala/collection/parallel/immutable/ParRange.scala index 350e64739f..64e07ce4ff 100644 --- a/src/library/scala/collection/parallel/immutable/ParRange.scala +++ b/src/library/scala/collection/parallel/immutable/ParRange.scala @@ -10,6 +10,7 @@ package scala.collection.parallel.immutable import scala.collection.immutable.Range import scala.collection.parallel.Combiner +import scala.collection.parallel.SeqSplitter import scala.collection.generic.CanCombineFrom import scala.collection.parallel.IterableSplitter import scala.collection.Iterator @@ -41,13 +42,10 @@ self => @inline final def apply(idx: Int) = range.apply(idx); - def splitter = new ParRangeIterator with SCPI - - type SCPI = SignalContextPassingIterator[ParRangeIterator] + def splitter = new ParRangeIterator class ParRangeIterator(range: Range = self.range) - extends ParIterator { - me: SignalContextPassingIterator[ParRangeIterator] => + extends SeqSplitter[Int] { override def toString = "ParRangeIterator(over: " + range + ")" private var ind = 0 private val len = range.length @@ -64,15 +62,15 @@ self => private def rangeleft = range.drop(ind) - def dup = new ParRangeIterator(rangeleft) with SCPI + def dup = new ParRangeIterator(rangeleft) def split = { val rleft = rangeleft val elemleft = rleft.length - if (elemleft < 2) Seq(new ParRangeIterator(rleft) with SCPI) + if (elemleft < 2) Seq(new ParRangeIterator(rleft)) else Seq( - new ParRangeIterator(rleft.take(elemleft / 2)) with SCPI, - new ParRangeIterator(rleft.drop(elemleft / 2)) with SCPI + new ParRangeIterator(rleft.take(elemleft / 2)), + new ParRangeIterator(rleft.drop(elemleft / 2)) ) } @@ -81,7 +79,7 @@ self => for (sz <- sizes) yield { val fronttaken = rleft.take(sz) rleft = rleft.drop(sz) - new ParRangeIterator(fronttaken) with SCPI + new ParRangeIterator(fronttaken) } } diff --git a/src/library/scala/collection/parallel/immutable/ParVector.scala b/src/library/scala/collection/parallel/immutable/ParVector.scala index fdeaefc3ff..5d9c431bc1 100644 --- a/src/library/scala/collection/parallel/immutable/ParVector.scala +++ b/src/library/scala/collection/parallel/immutable/ParVector.scala @@ -48,22 +48,19 @@ extends ParSeq[T] def this() = this(Vector()) - type SCPI = SignalContextPassingIterator[ParVectorIterator] - def apply(idx: Int) = vector.apply(idx) def length = vector.length def splitter: SeqSplitter[T] = { - val pit = new ParVectorIterator(vector.startIndex, vector.endIndex) with SCPI + val pit = new ParVectorIterator(vector.startIndex, vector.endIndex) vector.initIterator(pit) pit } override def seq: Vector[T] = vector - class ParVectorIterator(_start: Int, _end: Int) extends VectorIterator[T](_start, _end) with ParIterator { - self: SCPI => + class ParVectorIterator(_start: Int, _end: Int) extends VectorIterator[T](_start, _end) with SeqSplitter[T] { def remaining: Int = remainingElementCount def dup: SeqSplitter[T] = (new ParVector(remainingVector)).splitter def split: Seq[ParVectorIterator] = { diff --git a/src/library/scala/collection/parallel/immutable/package.scala b/src/library/scala/collection/parallel/immutable/package.scala index 7b1e39d092..63635537d7 100644 --- a/src/library/scala/collection/parallel/immutable/package.scala +++ b/src/library/scala/collection/parallel/immutable/package.scala @@ -22,23 +22,19 @@ package immutable { override def seq = throw new UnsupportedOperationException def update(idx: Int, elem: T) = throw new UnsupportedOperationException - type SCPI = SignalContextPassingIterator[ParIterator] - - class ParIterator(var i: Int = 0, val until: Int = length, elem: T = self.elem) extends super.ParIterator { - me: SignalContextPassingIterator[ParIterator] => - + class ParIterator(var i: Int = 0, val until: Int = length, elem: T = self.elem) extends SeqSplitter[T] { def remaining = until - i def hasNext = i < until def next = { i += 1; elem } - def dup = new ParIterator(i, until, elem) with SCPI + def dup = new ParIterator(i, until, elem) def psplit(sizes: Int*) = { val incr = sizes.scanLeft(0)(_ + _) - for ((start, end) <- incr.init zip incr.tail) yield new ParIterator(i + start, (i + end) min until, elem) with SCPI + for ((start, end) <- incr.init zip incr.tail) yield new ParIterator(i + start, (i + end) min until, elem) } def split = psplit(remaining / 2, remaining - remaining / 2) } - def splitter = new ParIterator with SCPI + def splitter = new ParIterator } } diff --git a/src/library/scala/collection/parallel/mutable/ParArray.scala b/src/library/scala/collection/parallel/mutable/ParArray.scala index a1eb3beb0c..72a8184b10 100644 --- a/src/library/scala/collection/parallel/mutable/ParArray.scala +++ b/src/library/scala/collection/parallel/mutable/ParArray.scala @@ -19,6 +19,7 @@ import scala.collection.generic.CanBuildFrom import scala.collection.generic.ParFactory import scala.collection.generic.Sizing import scala.collection.parallel.Combiner +import scala.collection.parallel.SeqSplitter import scala.collection.parallel.ParSeqLike import scala.collection.parallel.CHECK_RATE import scala.collection.mutable.ArraySeq @@ -74,17 +75,13 @@ self => override def seq = arrayseq - type SCPI = SignalContextPassingIterator[ParArrayIterator] - protected[parallel] def splitter: ParArrayIterator = { - val pit = new ParArrayIterator with SCPI + val pit = new ParArrayIterator pit } class ParArrayIterator(var i: Int = 0, val until: Int = length, val arr: Array[Any] = array) - extends super.ParIterator { - me: SignalContextPassingIterator[ParArrayIterator] => - + extends SeqSplitter[T] { def hasNext = i < until def next = { @@ -95,9 +92,9 @@ self => def remaining = until - i - def dup = new ParArrayIterator(i, until, arr) with SCPI + def dup = new ParArrayIterator(i, until, arr) - def psplit(sizesIncomplete: Int*): Seq[ParIterator] = { + def psplit(sizesIncomplete: Int*): Seq[ParArrayIterator] = { var traversed = i val total = sizesIncomplete.reduceLeft(_ + _) val left = remaining @@ -106,19 +103,19 @@ self => val start = traversed val end = (traversed + sz) min until traversed = end - new ParArrayIterator(start, end, arr) with SCPI + new ParArrayIterator(start, end, arr) } else { - new ParArrayIterator(traversed, traversed, arr) with SCPI + new ParArrayIterator(traversed, traversed, arr) } } - override def split: Seq[ParIterator] = { + override def split: Seq[ParArrayIterator] = { val left = remaining if (left >= 2) { val splitpoint = left / 2 val sq = Seq( - new ParArrayIterator(i, i + splitpoint, arr) with SCPI, - new ParArrayIterator(i + splitpoint, until, arr) with SCPI) + new ParArrayIterator(i, i + splitpoint, arr), + new ParArrayIterator(i + splitpoint, until, arr)) i = until sq } else { diff --git a/src/library/scala/collection/parallel/mutable/ParHashMap.scala b/src/library/scala/collection/parallel/mutable/ParHashMap.scala index 31750b0b0d..3b4d3dc0b0 100644 --- a/src/library/scala/collection/parallel/mutable/ParHashMap.scala +++ b/src/library/scala/collection/parallel/mutable/ParHashMap.scala @@ -56,7 +56,7 @@ self => override def seq = new collection.mutable.HashMap[K, V](hashTableContents) - def splitter = new ParHashMapIterator(1, table.length, size, table(0).asInstanceOf[DefaultEntry[K, V]]) with SCPI + def splitter = new ParHashMapIterator(1, table.length, size, table(0).asInstanceOf[DefaultEntry[K, V]]) override def size = tableSize @@ -93,14 +93,11 @@ self => override def stringPrefix = "ParHashMap" - type SCPI = SignalContextPassingIterator[ParHashMapIterator] - class ParHashMapIterator(start: Int, untilIdx: Int, totalSize: Int, e: DefaultEntry[K, V]) - extends EntryIterator[(K, V), ParHashMapIterator](start, untilIdx, totalSize, e) with ParIterator { - me: SCPI => + extends EntryIterator[(K, V), ParHashMapIterator](start, untilIdx, totalSize, e) { def entry2item(entry: DefaultEntry[K, V]) = (entry.key, entry.value); def newIterator(idxFrom: Int, idxUntil: Int, totalSz: Int, es: DefaultEntry[K, V]) = - new ParHashMapIterator(idxFrom, idxUntil, totalSz, es) with SCPI + new ParHashMapIterator(idxFrom, idxUntil, totalSz, es) } private def writeObject(out: java.io.ObjectOutputStream) { diff --git a/src/library/scala/collection/parallel/mutable/ParHashSet.scala b/src/library/scala/collection/parallel/mutable/ParHashSet.scala index 7763cdf318..6c5f513ad0 100644 --- a/src/library/scala/collection/parallel/mutable/ParHashSet.scala +++ b/src/library/scala/collection/parallel/mutable/ParHashSet.scala @@ -66,14 +66,11 @@ extends ParSet[T] def contains(elem: T) = containsEntry(elem) - def splitter = new ParHashSetIterator(0, table.length, size) with SCPI - - type SCPI = SignalContextPassingIterator[ParHashSetIterator] + def splitter = new ParHashSetIterator(0, table.length, size) class ParHashSetIterator(start: Int, iteratesUntil: Int, totalElements: Int) - extends ParFlatHashTableIterator(start, iteratesUntil, totalElements) with ParIterator { - me: SCPI => - def newIterator(start: Int, until: Int, total: Int) = new ParHashSetIterator(start, until, total) with SCPI + extends ParFlatHashTableIterator(start, iteratesUntil, totalElements) { + def newIterator(start: Int, until: Int, total: Int) = new ParHashSetIterator(start, until, total) } private def writeObject(s: java.io.ObjectOutputStream) { diff --git a/src/library/scala/collection/parallel/mutable/ParHashTable.scala b/src/library/scala/collection/parallel/mutable/ParHashTable.scala index 9b8e233b95..8c93732427 100644 --- a/src/library/scala/collection/parallel/mutable/ParHashTable.scala +++ b/src/library/scala/collection/parallel/mutable/ParHashTable.scala @@ -29,7 +29,7 @@ trait ParHashTable[K, Entry >: Null <: HashEntry[K, Entry]] extends collection.m /** A parallel iterator returning all the entries. */ abstract class EntryIterator[T, +IterRepr <: IterableSplitter[T]] - (private var idx: Int, private val until: Int, private val totalsize: Int, private var es: Entry) + (private var idx: Int, private val until: Int, private val totalsize: Int, private var es: Entry) extends IterableSplitter[T] with SizeMapUtils { private val itertable = table private var traversed = 0 diff --git a/src/library/scala/collection/parallel/package.scala b/src/library/scala/collection/parallel/package.scala index f154019bac..cdb9944fdc 100644 --- a/src/library/scala/collection/parallel/package.scala +++ b/src/library/scala/collection/parallel/package.scala @@ -140,8 +140,9 @@ package parallel { * Automatically forwards the signal delegate when splitting. */ private[parallel] class BufferSplitter[T] - (private val buffer: collection.mutable.ArrayBuffer[T], private var index: Int, private val until: Int, var signalDelegate: collection.generic.Signalling) + (private val buffer: collection.mutable.ArrayBuffer[T], private var index: Int, private val until: Int, _sigdel: collection.generic.Signalling) extends IterableSplitter[T] { + signalDelegate = _sigdel def hasNext = index < until def next = { val r = buffer(index) |