diff options
Diffstat (limited to 'src/library')
18 files changed, 401 insertions, 274 deletions
diff --git a/src/library/scala/collection/parallel/Combiner.scala b/src/library/scala/collection/parallel/Combiner.scala index 54cdf25804..6afe901258 100644 --- a/src/library/scala/collection/parallel/Combiner.scala +++ b/src/library/scala/collection/parallel/Combiner.scala @@ -33,8 +33,21 @@ import scala.collection.generic.Sizing * @since 2.9 */ trait Combiner[-Elem, +To] extends Builder[Elem, To] with Sizing with Parallel { -//self: EnvironmentPassingCombiner[Elem, To] => - + + @transient + @volatile + var _combinerTaskSupport = defaultTaskSupport + + def combinerTaskSupport = { + val cts = _combinerTaskSupport + if (cts eq null) { + _combinerTaskSupport = defaultTaskSupport + defaultTaskSupport + } else cts + } + + def combinerTaskSupport_=(cts: TaskSupport) = _combinerTaskSupport = cts + /** Combines the contents of the receiver builder and the `other` builder, * producing a new builder containing both their elements. * @@ -68,7 +81,15 @@ trait Combiner[-Elem, +To] extends Builder[Elem, To] with Sizing with Parallel { * By default, this method returns `false`. */ def canBeShared: Boolean = false - + + /** Constructs the result and sets the appropriate tasksupport object to the resulting collection + * if this is applicable. + */ + def resultWithTaskSupport: To = { + val res = result + setTaskSupport(res, combinerTaskSupport) + } + } diff --git a/src/library/scala/collection/parallel/ParIterableLike.scala b/src/library/scala/collection/parallel/ParIterableLike.scala index c0fc906ad9..5e6bf8c1a3 100644 --- a/src/library/scala/collection/parallel/ParIterableLike.scala +++ b/src/library/scala/collection/parallel/ParIterableLike.scala @@ -28,7 +28,7 @@ import immutable.HashMapCombiner import java.util.concurrent.atomic.AtomicBoolean import annotation.unchecked.uncheckedVariance - +import annotation.unchecked.uncheckedStable /** A template trait for parallel collections of type `ParIterable[T]`. @@ -154,9 +154,21 @@ extends GenIterableLike[T, Repr] with HasNewCombiner[T, Repr] { self: ParIterableLike[T, Repr, Sequential] => - - import tasksupport._ - + + @transient + @volatile + private var _tasksupport = defaultTaskSupport + + def tasksupport = { + val ts = _tasksupport + if (ts eq null) { + _tasksupport = defaultTaskSupport + defaultTaskSupport + } else ts + } + + def tasksupport_=(ts: TaskSupport) = _tasksupport = ts + def seq: Sequential def repr: Repr = this.asInstanceOf[Repr] @@ -306,7 +318,7 @@ self: ParIterableLike[T, Repr, Sequential] => * if this $coll is empty. */ def reduce[U >: T](op: (U, U) => U): U = { - executeAndWaitResult(new Reduce(op, splitter) mapResult { _.get }) + tasksupport.executeAndWaitResult(new Reduce(op, splitter) mapResult { _.get }) } /** Optionally reduces the elements of this sequence using the specified associative binary operator. @@ -341,7 +353,7 @@ self: ParIterableLike[T, Repr, Sequential] => * @return the result of applying fold operator `op` between all the elements and `z` */ def fold[U >: T](z: U)(op: (U, U) => U): U = { - executeAndWaitResult(new Fold(z, op, splitter)) + tasksupport.executeAndWaitResult(new Fold(z, op, splitter)) } /** Aggregates the results of applying an operator to subsequent elements. @@ -373,7 +385,7 @@ self: ParIterableLike[T, Repr, Sequential] => * @param combop an associative operator used to combine results from different partitions */ def aggregate[S](z: S)(seqop: (S, T) => S, combop: (S, S) => S): S = { - executeAndWaitResult(new Aggregate(z, seqop, combop, splitter)) + tasksupport.executeAndWaitResult(new Aggregate(z, seqop, combop, splitter)) } def foldLeft[S](z: S)(op: (S, T) => S): S = seq.foldLeft(z)(op) @@ -394,27 +406,27 @@ self: ParIterableLike[T, Repr, Sequential] => * @param f function applied to each element */ def foreach[U](f: T => U) = { - executeAndWaitResult(new Foreach(f, splitter)) + tasksupport.executeAndWaitResult(new Foreach(f, splitter)) } def count(p: T => Boolean): Int = { - executeAndWaitResult(new Count(p, splitter)) + tasksupport.executeAndWaitResult(new Count(p, splitter)) } def sum[U >: T](implicit num: Numeric[U]): U = { - executeAndWaitResult(new Sum[U](num, splitter)) + tasksupport.executeAndWaitResult(new Sum[U](num, splitter)) } def product[U >: T](implicit num: Numeric[U]): U = { - executeAndWaitResult(new Product[U](num, splitter)) + tasksupport.executeAndWaitResult(new Product[U](num, splitter)) } def min[U >: T](implicit ord: Ordering[U]): T = { - executeAndWaitResult(new Min(ord, splitter) mapResult { _.get }).asInstanceOf[T] + tasksupport.executeAndWaitResult(new Min(ord, splitter) mapResult { _.get }).asInstanceOf[T] } def max[U >: T](implicit ord: Ordering[U]): T = { - executeAndWaitResult(new Max(ord, splitter) mapResult { _.get }).asInstanceOf[T] + tasksupport.executeAndWaitResult(new Max(ord, splitter) mapResult { _.get }).asInstanceOf[T] } def maxBy[S](f: T => S)(implicit cmp: Ordering[S]): T = { @@ -430,24 +442,24 @@ self: ParIterableLike[T, Repr, Sequential] => } def map[S, That](f: T => S)(implicit bf: CanBuildFrom[Repr, S, That]): That = if (bf(repr).isCombiner) { - executeAndWaitResult(new Map[S, That](f, combinerFactory(() => bf(repr).asCombiner), splitter) mapResult { _.result }) - } else seq.map(f)(bf2seq(bf)) + tasksupport.executeAndWaitResult(new Map[S, That](f, combinerFactory(() => bf(repr).asCombiner), splitter) mapResult { _.resultWithTaskSupport }) + } else setTaskSupport(seq.map(f)(bf2seq(bf)), tasksupport) /*bf ifParallel { pbf => - executeAndWaitResult(new Map[S, That](f, pbf, splitter) mapResult { _.result }) + tasksupport.executeAndWaitResult(new Map[S, That](f, pbf, splitter) mapResult { _.result }) } otherwise seq.map(f)(bf2seq(bf))*/ def collect[S, That](pf: PartialFunction[T, S])(implicit bf: CanBuildFrom[Repr, S, That]): That = if (bf(repr).isCombiner) { - executeAndWaitResult(new Collect[S, That](pf, combinerFactory(() => bf(repr).asCombiner), splitter) mapResult { _.result }) - } else seq.collect(pf)(bf2seq(bf)) + tasksupport.executeAndWaitResult(new Collect[S, That](pf, combinerFactory(() => bf(repr).asCombiner), splitter) mapResult { _.resultWithTaskSupport }) + } else setTaskSupport(seq.collect(pf)(bf2seq(bf)), tasksupport) /*bf ifParallel { pbf => - executeAndWaitResult(new Collect[S, That](pf, pbf, splitter) mapResult { _.result }) + tasksupport.executeAndWaitResult(new Collect[S, That](pf, pbf, splitter) mapResult { _.result }) } otherwise seq.collect(pf)(bf2seq(bf))*/ def flatMap[S, That](f: T => GenTraversableOnce[S])(implicit bf: CanBuildFrom[Repr, S, That]): That = if (bf(repr).isCombiner) { - executeAndWaitResult(new FlatMap[S, That](f, combinerFactory(() => bf(repr).asCombiner), splitter) mapResult { _.result }) - } else seq.flatMap(f)(bf2seq(bf)) + tasksupport.executeAndWaitResult(new FlatMap[S, That](f, combinerFactory(() => bf(repr).asCombiner), splitter) mapResult { _.resultWithTaskSupport }) + } else setTaskSupport(seq.flatMap(f)(bf2seq(bf)), tasksupport) /*bf ifParallel { pbf => - executeAndWaitResult(new FlatMap[S, That](f, pbf, splitter) mapResult { _.result }) + tasksupport.executeAndWaitResult(new FlatMap[S, That](f, pbf, splitter) mapResult { _.result }) } otherwise seq.flatMap(f)(bf2seq(bf))*/ /** Tests whether a predicate holds for all elements of this $coll. @@ -458,7 +470,7 @@ self: ParIterableLike[T, Repr, Sequential] => * @return true if `p` holds for all elements, false otherwise */ def forall(pred: T => Boolean): Boolean = { - executeAndWaitResult(new Forall(pred, splitter assign new DefaultSignalling with VolatileAbort)) + tasksupport.executeAndWaitResult(new Forall(pred, splitter assign new DefaultSignalling with VolatileAbort)) } /** Tests whether a predicate holds for some element of this $coll. @@ -469,7 +481,7 @@ self: ParIterableLike[T, Repr, Sequential] => * @return true if `p` holds for some element, false otherwise */ def exists(pred: T => Boolean): Boolean = { - executeAndWaitResult(new Exists(pred, splitter assign new DefaultSignalling with VolatileAbort)) + tasksupport.executeAndWaitResult(new Exists(pred, splitter assign new DefaultSignalling with VolatileAbort)) } /** Finds some element in the collection for which the predicate holds, if such @@ -484,7 +496,7 @@ self: ParIterableLike[T, Repr, Sequential] => * @return an option value with the element if such an element exists, or `None` otherwise */ def find(pred: T => Boolean): Option[T] = { - executeAndWaitResult(new Find(pred, splitter assign new DefaultSignalling with VolatileAbort)) + tasksupport.executeAndWaitResult(new Find(pred, splitter assign new DefaultSignalling with VolatileAbort)) } /** Creates a combiner factory. Each combiner factory instance is used @@ -500,6 +512,7 @@ self: ParIterableLike[T, Repr, Sequential] => */ protected[this] def combinerFactory = { val combiner = newCombiner + combiner.combinerTaskSupport = tasksupport if (combiner.canBeShared) new CombinerFactory[T, Repr] { val shared = combiner def apply() = shared @@ -512,6 +525,7 @@ self: ParIterableLike[T, Repr, Sequential] => protected[this] def combinerFactory[S, That](cbf: () => Combiner[S, That]) = { val combiner = cbf() + combiner.combinerTaskSupport = tasksupport if (combiner.canBeShared) new CombinerFactory[S, That] { val shared = combiner def apply() = shared @@ -523,11 +537,11 @@ self: ParIterableLike[T, Repr, Sequential] => } def filter(pred: T => Boolean): Repr = { - executeAndWaitResult(new Filter(pred, combinerFactory, splitter) mapResult { _.result }) + tasksupport.executeAndWaitResult(new Filter(pred, combinerFactory, splitter) mapResult { _.resultWithTaskSupport }) } def filterNot(pred: T => Boolean): Repr = { - executeAndWaitResult(new FilterNot(pred, combinerFactory, splitter) mapResult { _.result }) + tasksupport.executeAndWaitResult(new FilterNot(pred, combinerFactory, splitter) mapResult { _.resultWithTaskSupport }) } def ++[U >: T, That](that: GenTraversableOnce[U])(implicit bf: CanBuildFrom[Repr, U, That]): That = { @@ -542,43 +556,47 @@ self: ParIterableLike[T, Repr, Sequential] => tasksupport.executeAndWaitResult(othtask) } val task = (copythis parallel copythat) { _ combine _ } mapResult { - _.result + _.resultWithTaskSupport } - executeAndWaitResult(task) - } else if (bf.isParallel) { + tasksupport.executeAndWaitResult(task) + } else if (bf(repr).isCombiner) { // println("case parallel builder, `that` not parallel") - val pbf = bf.asParallel - val copythis = new Copy(combinerFactory(() => pbf(repr)), splitter) + val copythis = new Copy(combinerFactory(() => bf(repr).asCombiner), splitter) val copythat = wrap { - val cb = pbf(repr) + val cb = bf(repr).asCombiner for (elem <- that.seq) cb += elem cb } - executeAndWaitResult((copythis parallel copythat) { _ combine _ } mapResult { _.result }) + tasksupport.executeAndWaitResult((copythis parallel copythat) { _ combine _ } mapResult { _.resultWithTaskSupport }) } else { // println("case not a parallel builder") val b = bf(repr) this.splitter.copy2builder[U, That, Builder[U, That]](b) for (elem <- that.seq) b += elem - b.result + setTaskSupport(b.result, tasksupport) } } def partition(pred: T => Boolean): (Repr, Repr) = { - executeAndWaitResult(new Partition(pred, combinerFactory, combinerFactory, splitter) mapResult { p => (p._1.result, p._2.result) }) + tasksupport.executeAndWaitResult( + new Partition(pred, combinerFactory, combinerFactory, splitter) mapResult { + p => (p._1.resultWithTaskSupport, p._2.resultWithTaskSupport) + } + ) } def groupBy[K](f: T => K): immutable.ParMap[K, Repr] = { - executeAndWaitResult(new GroupBy(f, () => HashMapCombiner[K, T], splitter) mapResult { + val r = tasksupport.executeAndWaitResult(new GroupBy(f, () => HashMapCombiner[K, T], splitter) mapResult { rcb => rcb.groupByKey(() => combinerFactory()) }) + setTaskSupport(r, tasksupport) } def take(n: Int): Repr = { val actualn = if (size > n) n else size if (actualn < MIN_FOR_COPY) take_sequential(actualn) - else executeAndWaitResult(new Take(actualn, combinerFactory, splitter) mapResult { - _.result + else tasksupport.executeAndWaitResult(new Take(actualn, combinerFactory, splitter) mapResult { + _.resultWithTaskSupport }) } @@ -591,13 +609,13 @@ self: ParIterableLike[T, Repr, Sequential] => cb += it.next left -= 1 } - cb.result + cb.resultWithTaskSupport } def drop(n: Int): Repr = { val actualn = if (size > n) n else size if ((size - actualn) < MIN_FOR_COPY) drop_sequential(actualn) - else executeAndWaitResult(new Drop(actualn, combinerFactory, splitter) mapResult { _.result }) + else tasksupport.executeAndWaitResult(new Drop(actualn, combinerFactory, splitter) mapResult { _.resultWithTaskSupport }) } private def drop_sequential(n: Int) = { @@ -605,14 +623,14 @@ self: ParIterableLike[T, Repr, Sequential] => val cb = newCombiner cb.sizeHint(size - n) while (it.hasNext) cb += it.next - cb.result + cb.resultWithTaskSupport } override def slice(unc_from: Int, unc_until: Int): Repr = { val from = unc_from min size max 0 val until = unc_until min size max from if ((until - from) <= MIN_FOR_COPY) slice_sequential(from, until) - else executeAndWaitResult(new Slice(from, until, combinerFactory, splitter) mapResult { _.result }) + else tasksupport.executeAndWaitResult(new Slice(from, until, combinerFactory, splitter) mapResult { _.resultWithTaskSupport }) } private def slice_sequential(from: Int, until: Int): Repr = { @@ -623,11 +641,15 @@ self: ParIterableLike[T, Repr, Sequential] => cb += it.next left -= 1 } - cb.result + cb.resultWithTaskSupport } def splitAt(n: Int): (Repr, Repr) = { - executeAndWaitResult(new SplitAt(n, combinerFactory, combinerFactory, splitter) mapResult { p => (p._1.result, p._2.result) }) + tasksupport.executeAndWaitResult( + new SplitAt(n, combinerFactory, combinerFactory, splitter) mapResult { + p => (p._1.resultWithTaskSupport, p._2.resultWithTaskSupport) + } + ) } /** Computes a prefix scan of the elements of the collection. @@ -645,20 +667,19 @@ self: ParIterableLike[T, Repr, Sequential] => * * @return a new $coll containing the prefix scan of the elements in this $coll */ - def scan[U >: T, That](z: U)(op: (U, U) => U)(implicit bf: CanBuildFrom[Repr, U, That]): That = if (bf.isParallel) { - val cbf = bf.asParallel - if (parallelismLevel > 1) { - if (size > 0) executeAndWaitResult(new CreateScanTree(0, size, z, op, splitter) mapResult { - tree => executeAndWaitResult(new FromScanTree(tree, z, op, combinerFactory(() => cbf(repr))) mapResult { - cb => cb.result + def scan[U >: T, That](z: U)(op: (U, U) => U)(implicit bf: CanBuildFrom[Repr, U, That]): That = if (bf(repr).isCombiner) { + if (tasksupport.parallelismLevel > 1) { + if (size > 0) tasksupport.executeAndWaitResult(new CreateScanTree(0, size, z, op, splitter) mapResult { + tree => tasksupport.executeAndWaitResult(new FromScanTree(tree, z, op, combinerFactory(() => bf(repr).asCombiner)) mapResult { + cb => cb.resultWithTaskSupport }) - }) else (cbf(self.repr) += z).result - } else seq.scan(z)(op)(bf2seq(bf)) - } else seq.scan(z)(op)(bf2seq(bf)) + }) else setTaskSupport((bf(repr) += z).result, tasksupport) + } else setTaskSupport(seq.scan(z)(op)(bf2seq(bf)), tasksupport) + } else setTaskSupport(seq.scan(z)(op)(bf2seq(bf)), tasksupport) - def scanLeft[S, That](z: S)(op: (S, T) => S)(implicit bf: CanBuildFrom[Repr, S, That]) = seq.scanLeft(z)(op)(bf2seq(bf)) + def scanLeft[S, That](z: S)(op: (S, T) => S)(implicit bf: CanBuildFrom[Repr, S, That]) = setTaskSupport(seq.scanLeft(z)(op)(bf2seq(bf)), tasksupport) - def scanRight[S, That](z: S)(op: (T, S) => S)(implicit bf: CanBuildFrom[Repr, S, That]) = seq.scanRight(z)(op)(bf2seq(bf)) + def scanRight[S, That](z: S)(op: (T, S) => S)(implicit bf: CanBuildFrom[Repr, S, That]) = setTaskSupport(seq.scanRight(z)(op)(bf2seq(bf)), tasksupport) /** Takes the longest prefix of elements that satisfy the predicate. * @@ -672,11 +693,15 @@ self: ParIterableLike[T, Repr, Sequential] => val cbf = combinerFactory if (cbf.doesShareCombiners) { val parseqspan = toSeq.takeWhile(pred) - executeAndWaitResult(new Copy(combinerFactory, parseqspan.splitter) mapResult { _.result }) + tasksupport.executeAndWaitResult(new Copy(combinerFactory, parseqspan.splitter) mapResult { + _.resultWithTaskSupport + }) } else { val cntx = new DefaultSignalling with AtomicIndexFlag cntx.setIndexFlag(Int.MaxValue) - executeAndWaitResult(new TakeWhile(0, pred, combinerFactory, splitter assign cntx) mapResult { _._1.result }) + tasksupport.executeAndWaitResult(new TakeWhile(0, pred, combinerFactory, splitter assign cntx) mapResult { + _._1.resultWithTaskSupport + }) } } @@ -693,17 +718,17 @@ self: ParIterableLike[T, Repr, Sequential] => val cbf = combinerFactory if (cbf.doesShareCombiners) { val (xs, ys) = toSeq.span(pred) - val copyxs = new Copy(combinerFactory, xs.splitter) mapResult { _.result } - val copyys = new Copy(combinerFactory, ys.splitter) mapResult { _.result } + val copyxs = new Copy(combinerFactory, xs.splitter) mapResult { _.resultWithTaskSupport } + val copyys = new Copy(combinerFactory, ys.splitter) mapResult { _.resultWithTaskSupport } val copyall = (copyxs parallel copyys) { (xr, yr) => (xr, yr) } - executeAndWaitResult(copyall) + tasksupport.executeAndWaitResult(copyall) } else { val cntx = new DefaultSignalling with AtomicIndexFlag cntx.setIndexFlag(Int.MaxValue) - executeAndWaitResult(new Span(0, pred, combinerFactory, combinerFactory, splitter assign cntx) mapResult { - p => (p._1.result, p._2.result) + tasksupport.executeAndWaitResult(new Span(0, pred, combinerFactory, combinerFactory, splitter assign cntx) mapResult { + p => (p._1.resultWithTaskSupport, p._2.resultWithTaskSupport) }) } } @@ -721,7 +746,11 @@ self: ParIterableLike[T, Repr, Sequential] => def dropWhile(pred: T => Boolean): Repr = { val cntx = new DefaultSignalling with AtomicIndexFlag cntx.setIndexFlag(Int.MaxValue) - executeAndWaitResult(new Span(0, pred, combinerFactory, combinerFactory, splitter assign cntx) mapResult { _._2.result }) + tasksupport.executeAndWaitResult( + new Span(0, pred, combinerFactory, combinerFactory, splitter assign cntx) mapResult { + _._2.resultWithTaskSupport + } + ) } def copyToArray[U >: T](xs: Array[U]) = copyToArray(xs, 0) @@ -729,31 +758,33 @@ self: ParIterableLike[T, Repr, Sequential] => def copyToArray[U >: T](xs: Array[U], start: Int) = copyToArray(xs, start, xs.length - start) def copyToArray[U >: T](xs: Array[U], start: Int, len: Int) = if (len > 0) { - executeAndWaitResult(new CopyToArray(start, len, xs, splitter)) + tasksupport.executeAndWaitResult(new CopyToArray(start, len, xs, splitter)) } def sameElements[U >: T](that: GenIterable[U]) = seq.sameElements(that) - def zip[U >: T, S, That](that: GenIterable[S])(implicit bf: CanBuildFrom[Repr, (U, S), That]): That = if (bf.isParallel && that.isParSeq) { - val pbf = bf.asParallel + def zip[U >: T, S, That](that: GenIterable[S])(implicit bf: CanBuildFrom[Repr, (U, S), That]): That = if (bf(repr).isCombiner && that.isParSeq) { val thatseq = that.asParSeq - executeAndWaitResult(new Zip(combinerFactory(() => pbf(repr)), splitter, thatseq.splitter) mapResult { _.result }); - } else seq.zip(that)(bf2seq(bf)) + tasksupport.executeAndWaitResult(new Zip(combinerFactory(() => bf(repr).asCombiner), splitter, thatseq.splitter) mapResult { _.resultWithTaskSupport }); + } else setTaskSupport(seq.zip(that)(bf2seq(bf)), tasksupport) def zipWithIndex[U >: T, That](implicit bf: CanBuildFrom[Repr, (U, Int), That]): That = this zip immutable.ParRange(0, size, 1, false) - def zipAll[S, U >: T, That](that: GenIterable[S], thisElem: U, thatElem: S)(implicit bf: CanBuildFrom[Repr, (U, S), That]): That = if (bf.isParallel && that.isParSeq) { - val pbf = bf.asParallel + def zipAll[S, U >: T, That](that: GenIterable[S], thisElem: U, thatElem: S)(implicit bf: CanBuildFrom[Repr, (U, S), That]): That = if (bf(repr).isCombiner && that.isParSeq) { val thatseq = that.asParSeq - executeAndWaitResult(new ZipAll(size max thatseq.length, thisElem, thatElem, combinerFactory(() => pbf(repr)), splitter, thatseq.splitter) mapResult { _.result }); - } else seq.zipAll(that, thisElem, thatElem)(bf2seq(bf)) + tasksupport.executeAndWaitResult( + new ZipAll(size max thatseq.length, thisElem, thatElem, combinerFactory(() => bf(repr).asCombiner), splitter, thatseq.splitter) mapResult { + _.resultWithTaskSupport + } + ); + } else setTaskSupport(seq.zipAll(that, thisElem, thatElem)(bf2seq(bf)), tasksupport) protected def toParCollection[U >: T, That](cbf: () => Combiner[U, That]): That = { - executeAndWaitResult(new ToParCollection(combinerFactory(cbf), splitter) mapResult { _.result }); + tasksupport.executeAndWaitResult(new ToParCollection(combinerFactory(cbf), splitter) mapResult { _.resultWithTaskSupport }); } protected def toParMap[K, V, That](cbf: () => Combiner[(K, V), That])(implicit ev: T <:< (K, V)): That = { - executeAndWaitResult(new ToParMap(combinerFactory(cbf), splitter)(ev) mapResult { _.result }) + tasksupport.executeAndWaitResult(new ToParMap(combinerFactory(cbf), splitter)(ev) mapResult { _.resultWithTaskSupport }) } def view = new ParIterableView[T, Repr, Sequential] { @@ -810,7 +841,7 @@ self: ParIterableLike[T, Repr, Sequential] => extends StrictSplitterCheckTask[R, Tp] { protected[this] val pit: IterableSplitter[T] protected[this] def newSubtask(p: IterableSplitter[T]): Accessor[R, Tp] - def shouldSplitFurther = pit.shouldSplitFurther(self.repr, parallelismLevel) + def shouldSplitFurther = pit.shouldSplitFurther(self.repr, tasksupport.parallelismLevel) 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 + ")" @@ -844,8 +875,8 @@ self: ParIterableLike[T, Repr, Sequential] => (f: First, s: Second) extends Composite[FR, SR, R, First, Second](f, s) { def leaf(prevr: Option[R]) = { - executeAndWaitResult(ft) - executeAndWaitResult(st) + tasksupport.executeAndWaitResult(ft) + tasksupport.executeAndWaitResult(st) mergeSubtasks } } @@ -855,8 +886,8 @@ self: ParIterableLike[T, Repr, Sequential] => (f: First, s: Second) extends Composite[FR, SR, R, First, Second](f, s) { def leaf(prevr: Option[R]) = { - val ftfuture = execute(ft) - executeAndWaitResult(st) + val ftfuture = tasksupport.execute(ft) + tasksupport.executeAndWaitResult(st) ftfuture() mergeSubtasks } @@ -867,7 +898,7 @@ self: ParIterableLike[T, Repr, Sequential] => @volatile var result: R1 = null.asInstanceOf[R1] def map(r: R): R1 def leaf(prevr: Option[R1]) = { - val initialResult = executeAndWaitResult(inner) + val initialResult = tasksupport.executeAndWaitResult(inner) result = map(initialResult) } private[parallel] override def signalAbort() { @@ -1339,7 +1370,7 @@ self: ParIterableLike[T, Repr, Sequential] => /* scan tree */ - protected[this] def scanBlockSize = (thresholdFromSize(size, parallelismLevel) / 2) max 1 + protected[this] def scanBlockSize = (thresholdFromSize(size, tasksupport.parallelismLevel) / 2) max 1 protected[this] trait ScanTree[U >: T] { def beginsAt: Int diff --git a/src/library/scala/collection/parallel/ParIterableViewLike.scala b/src/library/scala/collection/parallel/ParIterableViewLike.scala index 1d7659922c..536139c812 100644 --- a/src/library/scala/collection/parallel/ParIterableViewLike.scala +++ b/src/library/scala/collection/parallel/ParIterableViewLike.scala @@ -47,7 +47,6 @@ extends GenIterableView[T, Coll] with ParIterableLike[T, This, ThisSeq] { self => - import tasksupport._ override def foreach[U](f: T => U): Unit = super[ParIterableLike].foreach(f) override protected[this] def newCombiner: Combiner[T, This] = throw new UnsupportedOperationException(this + ".newCombiner"); @@ -135,7 +134,7 @@ self => newZippedAllTryParSeq(that, thisElem, thatElem).asInstanceOf[That] override def force[U >: T, That](implicit bf: CanBuildFrom[Coll, U, That]) = bf ifParallel { pbf => - executeAndWaitResult(new Force(pbf, splitter).mapResult(_.result).asInstanceOf[Task[That, ResultMapping[_, Force[U, That], That]]]) + tasksupport.executeAndWaitResult(new Force(pbf, splitter).mapResult(_.result).asInstanceOf[Task[That, ResultMapping[_, Force[U, That], That]]]) } otherwise { val b = bf(underlying) b ++= this.iterator diff --git a/src/library/scala/collection/parallel/ParSeqLike.scala b/src/library/scala/collection/parallel/ParSeqLike.scala index 70529229ec..9f28a286ca 100644 --- a/src/library/scala/collection/parallel/ParSeqLike.scala +++ b/src/library/scala/collection/parallel/ParSeqLike.scala @@ -44,8 +44,7 @@ trait ParSeqLike[+T, +Repr <: ParSeq[T], +Sequential <: Seq[T] with SeqLike[T, S extends scala.collection.GenSeqLike[T, Repr] with ParIterableLike[T, Repr, Sequential] { self => - import tasksupport._ - + type SuperParIterator = IterableSplitter[T] /** A more refined version of the iterator found in the `ParallelIterable` trait, @@ -107,7 +106,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.psplitWithSignalling(realfrom, length - realfrom)(1) assign ctx))._1 + tasksupport.executeAndWaitResult(new SegmentLength(p, 0, splitter.psplitWithSignalling(realfrom, length - realfrom)(1) assign ctx))._1 } /** Finds the first element satisfying some predicate. @@ -125,7 +124,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.psplitWithSignalling(realfrom, length - realfrom)(1) assign ctx)) + tasksupport.executeAndWaitResult(new IndexWhere(p, realfrom, splitter.psplitWithSignalling(realfrom, length - realfrom)(1) assign ctx)) } /** Finds the last element satisfying some predicate. @@ -143,18 +142,20 @@ 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.psplitWithSignalling(until, length - until)(0) assign ctx)) + tasksupport.executeAndWaitResult(new LastIndexWhere(p, 0, splitter.psplitWithSignalling(until, length - until)(0) assign ctx)) } def reverse: Repr = { - executeAndWaitResult(new Reverse(() => newCombiner, splitter) mapResult { _.result }) + tasksupport.executeAndWaitResult(new Reverse(() => newCombiner, splitter) mapResult { _.resultWithTaskSupport }) } def reverseMap[S, That](f: T => S)(implicit bf: CanBuildFrom[Repr, S, That]): That = if (bf(repr).isCombiner) { - executeAndWaitResult(new ReverseMap[S, That](f, () => bf(repr).asCombiner, splitter) mapResult { _.result }) - } else seq.reverseMap(f)(bf2seq(bf)) + tasksupport.executeAndWaitResult( + new ReverseMap[S, That](f, () => bf(repr).asCombiner, splitter) mapResult { _.resultWithTaskSupport } + ) + } else setTaskSupport(seq.reverseMap(f)(bf2seq(bf)), tasksupport) /*bf ifParallel { pbf => - executeAndWaitResult(new ReverseMap[S, That](f, pbf, splitter) mapResult { _.result }) + tasksupport.executeAndWaitResult(new ReverseMap[S, That](f, pbf, splitter) mapResult { _.result }) } otherwise seq.reverseMap(f)(bf2seq(bf))*/ /** Tests whether this $coll contains the given sequence at a given index. @@ -172,13 +173,15 @@ self => else if (pthat.length > length - offset) false else { val ctx = new DefaultSignalling with VolatileAbort - executeAndWaitResult(new SameElements(splitter.psplitWithSignalling(offset, pthat.length)(1) assign ctx, pthat.splitter)) + tasksupport.executeAndWaitResult( + new SameElements(splitter.psplitWithSignalling(offset, pthat.length)(1) assign ctx, pthat.splitter) + ) } } otherwise seq.startsWith(that, offset) override def sameElements[U >: T](that: GenIterable[U]): Boolean = that ifParSeq { pthat => val ctx = new DefaultSignalling with VolatileAbort - length == pthat.length && executeAndWaitResult(new SameElements(splitter assign ctx, pthat.splitter)) + length == pthat.length && tasksupport.executeAndWaitResult(new SameElements(splitter assign ctx, pthat.splitter)) } otherwise seq.sameElements(that) /** Tests whether this $coll ends with the given parallel sequence. @@ -195,25 +198,24 @@ self => else { val ctx = new DefaultSignalling with VolatileAbort val tlen = that.length - executeAndWaitResult(new SameElements(splitter.psplitWithSignalling(length - tlen, tlen)(1) assign ctx, pthat.splitter)) + tasksupport.executeAndWaitResult(new SameElements(splitter.psplitWithSignalling(length - tlen, tlen)(1) assign ctx, pthat.splitter)) } } otherwise seq.endsWith(that) def patch[U >: T, That](from: Int, patch: GenSeq[U], replaced: Int)(implicit bf: CanBuildFrom[Repr, U, That]): That = { val realreplaced = replaced min (length - from) - if (patch.isParSeq && bf.isParallel && (size - realreplaced + patch.size) > MIN_FOR_COPY) { + if (patch.isParSeq && bf(repr).isCombiner && (size - realreplaced + patch.size) > MIN_FOR_COPY) { val that = patch.asParSeq - val pbf = bf.asParallel val pits = splitter.psplitWithSignalling(from, replaced, length - from - realreplaced) - val cfactory = combinerFactory(() => pbf(repr)) + val cfactory = combinerFactory(() => bf(repr).asCombiner) val copystart = new Copy[U, That](cfactory, pits(0)) val copymiddle = wrap { val tsk = new that.Copy[U, That](cfactory, that.splitter) tasksupport.executeAndWaitResult(tsk) } val copyend = new Copy[U, That](cfactory, pits(2)) - executeAndWaitResult(((copystart parallel copymiddle) { _ combine _ } parallel copyend) { _ combine _ } mapResult { - _.result + tasksupport.executeAndWaitResult(((copystart parallel copymiddle) { _ combine _ } parallel copyend) { _ combine _ } mapResult { + _.resultWithTaskSupport }) } else patch_sequential(from, patch.seq, replaced) } @@ -226,14 +228,18 @@ self => b ++= pits(0) b ++= patch b ++= pits(2) - b.result + setTaskSupport(b.result, tasksupport) } def updated[U >: T, That](index: Int, elem: U)(implicit bf: CanBuildFrom[Repr, U, That]): That = if (bf(repr).isCombiner) { - executeAndWaitResult(new Updated(index, elem, () => bf(repr).asCombiner, splitter) mapResult { _.result }) - } else seq.updated(index, elem)(bf2seq(bf)) + tasksupport.executeAndWaitResult( + new Updated(index, elem, combinerFactory(() => bf(repr).asCombiner), splitter) mapResult { + _.resultWithTaskSupport + } + ) + } else setTaskSupport(seq.updated(index, elem)(bf2seq(bf)), tasksupport) /*bf ifParallel { pbf => - executeAndWaitResult(new Updated(index, elem, pbf, splitter) mapResult { _.result }) + tasksupport.executeAndWaitResult(new Updated(index, elem, pbf, splitter) mapResult { _.result }) } otherwise seq.updated(index, elem)(bf2seq(bf))*/ def +:[U >: T, That](elem: U)(implicit bf: CanBuildFrom[Repr, U, That]): That = { @@ -248,10 +254,13 @@ self => patch(length, new immutable.Repetition(elem, len - length), 0) } else patch(length, Nil, 0); - override def zip[U >: T, S, That](that: GenIterable[S])(implicit bf: CanBuildFrom[Repr, (U, S), That]): That = if (bf.isParallel && that.isParSeq) { - val pbf = bf.asParallel + override def zip[U >: T, S, That](that: GenIterable[S])(implicit bf: CanBuildFrom[Repr, (U, S), That]): That = if (bf(repr).isCombiner && that.isParSeq) { val thatseq = that.asParSeq - executeAndWaitResult(new Zip(length min thatseq.length, pbf, splitter, thatseq.splitter) mapResult { _.result }); + tasksupport.executeAndWaitResult( + new Zip(length min thatseq.length, combinerFactory(() => bf(repr).asCombiner), splitter, thatseq.splitter) mapResult { + _.resultWithTaskSupport + } + ); } else super.zip(that)(bf) /** Tests whether every element of this $coll relates to the @@ -268,7 +277,7 @@ self => */ def corresponds[S](that: GenSeq[S])(p: (T, S) => Boolean): Boolean = that ifParSeq { pthat => val ctx = new DefaultSignalling with VolatileAbort - length == pthat.length && executeAndWaitResult(new Corresponds(p, splitter assign ctx, pthat.splitter)) + length == pthat.length && tasksupport.executeAndWaitResult(new Corresponds(p, splitter assign ctx, pthat.splitter)) } otherwise seq.corresponds(that)(p) def diff[U >: T](that: GenSeq[U]): Repr = sequentially { @@ -424,7 +433,7 @@ self => override def requiresStrictSplitters = true } - protected[this] class Updated[U >: T, That](pos: Int, elem: U, pbf: () => Combiner[U, That], protected[this] val pit: SeqSplitter[T]) + protected[this] class Updated[U >: T, That](pos: Int, elem: U, pbf: CombinerFactory[U, That], protected[this] val pit: SeqSplitter[T]) extends Transformer[Combiner[U, That], Updated[U, That]] { @volatile var result: Combiner[U, That] = null def leaf(prev: Option[Combiner[U, That]]) = result = pit.updated2combiner(pos, elem, pbf()) @@ -437,10 +446,10 @@ self => override def requiresStrictSplitters = true } - protected[this] class Zip[U >: T, S, That](len: Int, pbf: CanCombineFrom[Repr, (U, S), That], protected[this] val pit: SeqSplitter[T], val otherpit: SeqSplitter[S]) + protected[this] class Zip[U >: T, S, That](len: Int, cf: CombinerFactory[(U, S), That], protected[this] val pit: SeqSplitter[T], val otherpit: SeqSplitter[S]) extends Transformer[Combiner[(U, S), That], Zip[U, S, That]] { @volatile var result: Result = null - def leaf(prev: Option[Result]) = result = pit.zip2combiner[U, S, That](otherpit, pbf(self.repr)) + def leaf(prev: Option[Result]) = result = pit.zip2combiner[U, S, That](otherpit, cf()) protected[this] def newSubtask(p: SuperParIterator) = unsupported override def split = { val fp = len / 2 @@ -448,8 +457,8 @@ self => 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)) + new Zip(fp, cf, pits(0), opits(0)), + new Zip(sp, cf, pits(1), opits(1)) ) } override def merge(that: Zip[U, S, That]) = result = result combine that.result diff --git a/src/library/scala/collection/parallel/ParSeqViewLike.scala b/src/library/scala/collection/parallel/ParSeqViewLike.scala index 6fdc181793..e0d1a7d6ff 100644 --- a/src/library/scala/collection/parallel/ParSeqViewLike.scala +++ b/src/library/scala/collection/parallel/ParSeqViewLike.scala @@ -38,7 +38,6 @@ extends GenSeqView[T, Coll] with ParSeqLike[T, This, ThisSeq] { self => - import tasksupport._ trait Transformed[+S] extends ParSeqView[S, Coll, CollSeq] with super[ParIterableView].Transformed[S] with super[GenSeqViewLike].Transformed[S] { @@ -170,7 +169,7 @@ self => override def scanRight[S, That](z: S)(op: (T, S) => S)(implicit bf: CanBuildFrom[This, S, That]): That = newForced(thisParSeq.scanRight(z)(op)).asInstanceOf[That] override def groupBy[K](f: T => K): immutable.ParMap[K, This] = thisParSeq.groupBy(f).map(kv => (kv._1, newForced(kv._2).asInstanceOf[This])) override def force[U >: T, That](implicit bf: CanBuildFrom[Coll, U, That]) = bf ifParallel { pbf => - executeAndWaitResult(new Force(pbf, splitter).mapResult(_.result).asInstanceOf[Task[That, _]]) + tasksupport.executeAndWaitResult(new Force(pbf, splitter).mapResult(_.result).asInstanceOf[Task[That, _]]) } otherwise { val b = bf(underlying) b ++= this.iterator diff --git a/src/library/scala/collection/parallel/TaskSupport.scala b/src/library/scala/collection/parallel/TaskSupport.scala index 20800250b4..fc99347316 100644 --- a/src/library/scala/collection/parallel/TaskSupport.scala +++ b/src/library/scala/collection/parallel/TaskSupport.scala @@ -11,15 +11,26 @@ package scala.collection.parallel - +import java.util.concurrent.ThreadPoolExecutor +import scala.concurrent.forkjoin.ForkJoinPool +import scala.concurrent.ExecutionContext trait TaskSupport extends Tasks -private[collection] class ForkJoinTaskSupport extends TaskSupport with AdaptiveWorkStealingForkJoinTasks -private[collection] class ThreadPoolTaskSupport extends TaskSupport with AdaptiveWorkStealingThreadPoolTasks +private[collection] class ForkJoinTaskSupport(val environment: ForkJoinPool = ForkJoinTasks.defaultForkJoinPool) +extends TaskSupport with AdaptiveWorkStealingForkJoinTasks + + +private[collection] class ThreadPoolTaskSupport(val environment: ThreadPoolExecutor = ThreadPoolTasks.defaultThreadPool) +extends TaskSupport with AdaptiveWorkStealingThreadPoolTasks + + +private[collection] class ExecutionContextTaskSupport(val environment: ExecutionContext = scala.concurrent.executionContext) +extends TaskSupport with ExecutionContextTasks + diff --git a/src/library/scala/collection/parallel/Tasks.scala b/src/library/scala/collection/parallel/Tasks.scala index e643444638..60a8bb1ed6 100644 --- a/src/library/scala/collection/parallel/Tasks.scala +++ b/src/library/scala/collection/parallel/Tasks.scala @@ -6,109 +6,109 @@ ** |/ ** \* */ - package scala.collection.parallel +import java.util.concurrent.ThreadPoolExecutor import scala.concurrent.forkjoin._ +import scala.concurrent.ExecutionContext import scala.util.control.Breaks._ - import annotation.unchecked.uncheckedVariance +trait Task[R, +Tp] { + type Result = R -/** A trait that declares task execution capabilities used - * by parallel collections. - */ -trait Tasks { - - private[parallel] val debugMessages = collection.mutable.ArrayBuffer[String]() - - private[parallel] def debuglog(s: String) = synchronized { - debugMessages += s - } + def repr = this.asInstanceOf[Tp] - trait Task[R, +Tp] { - type Result = R - - def repr = this.asInstanceOf[Tp] - - /** Body of the task - non-divisible unit of work done by this task. - * Optionally is provided with the result from the previous completed task - * or `None` if there was no previous task (or the previous task is uncompleted or unknown). - */ - def leaf(result: Option[R]) + /** Body of the task - non-divisible unit of work done by this task. + * Optionally is provided with the result from the previous completed task + * or `None` if there was no previous task (or the previous task is uncompleted or unknown). + */ + def leaf(result: Option[R]) - /** A result that can be accessed once the task is completed. */ - var result: R + /** A result that can be accessed once the task is completed. */ + var result: R - /** Decides whether or not this task should be split further. */ - def shouldSplitFurther: Boolean + /** Decides whether or not this task should be split further. */ + def shouldSplitFurther: Boolean - /** Splits this task into a list of smaller tasks. */ - private[parallel] def split: Seq[Task[R, Tp]] + /** Splits this task into a list of smaller tasks. */ + private[parallel] def split: Seq[Task[R, Tp]] - /** Read of results of `that` task and merge them into results of this one. */ - private[parallel] def merge(that: Tp @uncheckedVariance) {} + /** Read of results of `that` task and merge them into results of this one. */ + private[parallel] def merge(that: Tp @uncheckedVariance) {} - // exception handling mechanism - @volatile var throwable: Throwable = null - def forwardThrowable() = if (throwable != null) throw throwable + // exception handling mechanism + @volatile var throwable: Throwable = null + def forwardThrowable() = if (throwable != null) throw throwable - // tries to do the leaf computation, storing the possible exception - private[parallel] def tryLeaf(lastres: Option[R]) { - try { - tryBreakable { - leaf(lastres) - result = result // ensure that effects of `leaf` are visible to readers of `result` - } catchBreak { - signalAbort - } - } catch { - case thr: Exception => - result = result // ensure that effects of `leaf` are visible - throwable = thr - signalAbort + // tries to do the leaf computation, storing the possible exception + private[parallel] def tryLeaf(lastres: Option[R]) { + try { + tryBreakable { + leaf(lastres) + result = result // ensure that effects of `leaf` are visible to readers of `result` + } catchBreak { + signalAbort } + } catch { + case thr: Exception => + result = result // ensure that effects of `leaf` are visible + throwable = thr + signalAbort } + } - private[parallel] def tryMerge(t: Tp @uncheckedVariance) { - val that = t.asInstanceOf[Task[R, Tp]] - val local = result // ensure that any effects of modifying `result` are detected - // checkMerge(that) - if (this.throwable == null && that.throwable == null) merge(t) - mergeThrowables(that) - } + private[parallel] def tryMerge(t: Tp @uncheckedVariance) { + val that = t.asInstanceOf[Task[R, Tp]] + val local = result // ensure that any effects of modifying `result` are detected + // checkMerge(that) + if (this.throwable == null && that.throwable == null) merge(t) + mergeThrowables(that) + } - private def checkMerge(that: Task[R, Tp] @uncheckedVariance) { - if (this.throwable == null && that.throwable == null && (this.result == null || that.result == null)) { - println("This: " + this + ", thr=" + this.throwable + "; merged with " + that + ", thr=" + that.throwable) - } else if (this.throwable != null || that.throwable != null) { - println("merging this: " + this + " with thr: " + this.throwable + " with " + that + ", thr=" + that.throwable) - } + private def checkMerge(that: Task[R, Tp] @uncheckedVariance) { + if (this.throwable == null && that.throwable == null && (this.result == null || that.result == null)) { + println("This: " + this + ", thr=" + this.throwable + "; merged with " + that + ", thr=" + that.throwable) + } else if (this.throwable != null || that.throwable != null) { + println("merging this: " + this + " with thr: " + this.throwable + " with " + that + ", thr=" + that.throwable) } + } - private[parallel] def mergeThrowables(that: Task[_, _]) { - if (this.throwable != null && that.throwable != null) { - // merge exceptions, since there were multiple exceptions - this.throwable = this.throwable alongWith that.throwable - } else if (that.throwable != null) this.throwable = that.throwable + private[parallel] def mergeThrowables(that: Task[_, _]) { + if (this.throwable != null && that.throwable != null) { + // merge exceptions, since there were multiple exceptions + this.throwable = this.throwable alongWith that.throwable + } else if (that.throwable != null) this.throwable = that.throwable else this.throwable = this.throwable - } + } + + // override in concrete task implementations to signal abort to other tasks + private[parallel] def signalAbort() {} +} + + +/** A trait that declares task execution capabilities used + * by parallel collections. + */ +trait Tasks { + + private[parallel] val debugMessages = collection.mutable.ArrayBuffer[String]() - // override in concrete task implementations to signal abort to other tasks - private[parallel] def signalAbort() {} + private[parallel] def debuglog(s: String) = synchronized { + debugMessages += s } - trait TaskImpl[R, +Tp] { + trait WrappedTask[R, +Tp] { /** the body of this task - what it executes, how it gets split and how results are merged. */ val body: Task[R, Tp] - def split: Seq[TaskImpl[R, Tp]] + def split: Seq[WrappedTask[R, Tp]] /** Code that gets called after the task gets started - it may spawn other tasks instead of calling `leaf`. */ def compute() /** Start task. */ @@ -129,13 +129,10 @@ trait Tasks { def release() {} } - protected def newTaskImpl[R, Tp](b: Task[R, Tp]): TaskImpl[R, Tp] - /* task control */ - // safe to assume it will always have the same type, - // because the `tasksupport` in parallel iterable is final - var environment: AnyRef + /** The type of the environment is more specific in the implementations. */ + val environment: AnyRef /** Executes a task and returns a future. Forwards an exception if some task threw it. */ def execute[R, Tp](fjtask: Task[R, Tp]): () => R @@ -155,11 +152,11 @@ trait Tasks { */ trait AdaptiveWorkStealingTasks extends Tasks { - trait TaskImpl[R, Tp] extends super.TaskImpl[R, Tp] { - @volatile var next: TaskImpl[R, Tp] = null + trait WrappedTask[R, Tp] extends super.WrappedTask[R, Tp] { + @volatile var next: WrappedTask[R, Tp] = null @volatile var shouldWaitFor = true - def split: Seq[TaskImpl[R, Tp]] + def split: Seq[WrappedTask[R, Tp]] def compute() = if (body.shouldSplitFurther) { internal() @@ -195,8 +192,8 @@ trait AdaptiveWorkStealingTasks extends Tasks { } def spawnSubtasks() = { - var last: TaskImpl[R, Tp] = null - var head: TaskImpl[R, Tp] = this + var last: WrappedTask[R, Tp] = null + var head: WrappedTask[R, Tp] = this do { val subtasks = head.split head = subtasks.head @@ -222,7 +219,7 @@ trait AdaptiveWorkStealingTasks extends Tasks { } // specialize ctor - protected def newTaskImpl[R, Tp](b: Task[R, Tp]): TaskImpl[R, Tp] + protected def newWrappedTask[R, Tp](b: Task[R, Tp]): WrappedTask[R, Tp] } @@ -231,7 +228,7 @@ trait AdaptiveWorkStealingTasks extends Tasks { trait ThreadPoolTasks extends Tasks { import java.util.concurrent._ - trait TaskImpl[R, +Tp] extends Runnable with super.TaskImpl[R, Tp] { + trait WrappedTask[R, +Tp] extends Runnable with super.WrappedTask[R, Tp] { // initially, this is null // once the task is started, this future is set and used for `sync` // utb: var future: Future[_] = null @@ -293,9 +290,9 @@ trait ThreadPoolTasks extends Tasks { } } - protected def newTaskImpl[R, Tp](b: Task[R, Tp]): TaskImpl[R, Tp] + protected def newWrappedTask[R, Tp](b: Task[R, Tp]): WrappedTask[R, Tp] - var environment: AnyRef = ThreadPoolTasks.defaultThreadPool + val environment: ThreadPoolExecutor def executor = environment.asInstanceOf[ThreadPoolExecutor] def queue = executor.getQueue.asInstanceOf[LinkedBlockingQueue[Runnable]] @volatile var totaltasks = 0 @@ -309,7 +306,7 @@ trait ThreadPoolTasks extends Tasks { } def execute[R, Tp](task: Task[R, Tp]): () => R = { - val t = newTaskImpl(task) + val t = newWrappedTask(task) // debuglog("-----------> Executing without wait: " + task) t.start() @@ -322,7 +319,7 @@ trait ThreadPoolTasks extends Tasks { } def executeAndWaitResult[R, Tp](task: Task[R, Tp]): R = { - val t = newTaskImpl(task) + val t = newWrappedTask(task) // debuglog("-----------> Executing with wait: " + task) t.start() @@ -362,10 +359,11 @@ object ThreadPoolTasks { /** An implementation of tasks objects based on the Java thread pooling API and synchronization using futures. */ +@deprecated("This implementation is not used.") trait FutureThreadPoolTasks extends Tasks { import java.util.concurrent._ - trait TaskImpl[R, +Tp] extends Runnable with super.TaskImpl[R, Tp] { + trait WrappedTask[R, +Tp] extends Runnable with super.WrappedTask[R, Tp] { @volatile var future: Future[_] = null def start() = { @@ -380,13 +378,13 @@ trait FutureThreadPoolTasks extends Tasks { } } - protected def newTaskImpl[R, Tp](b: Task[R, Tp]): TaskImpl[R, Tp] + protected def newWrappedTask[R, Tp](b: Task[R, Tp]): WrappedTask[R, Tp] - var environment: AnyRef = FutureThreadPoolTasks.defaultThreadPool + val environment: AnyRef = FutureThreadPoolTasks.defaultThreadPool def executor = environment.asInstanceOf[ThreadPoolExecutor] def execute[R, Tp](task: Task[R, Tp]): () => R = { - val t = newTaskImpl(task) + val t = newWrappedTask(task) // debuglog("-----------> Executing without wait: " + task) t.start @@ -399,7 +397,7 @@ trait FutureThreadPoolTasks extends Tasks { } def executeAndWaitResult[R, Tp](task: Task[R, Tp]): R = { - val t = newTaskImpl(task) + val t = newWrappedTask(task) // debuglog("-----------> Executing with wait: " + task) t.start @@ -441,26 +439,26 @@ trait HavingForkJoinPool { */ trait ForkJoinTasks extends Tasks with HavingForkJoinPool { - trait TaskImpl[R, +Tp] extends RecursiveAction with super.TaskImpl[R, Tp] { + trait WrappedTask[R, +Tp] extends RecursiveAction with super.WrappedTask[R, Tp] { def start() = fork def sync() = join def tryCancel = tryUnfork } // specialize ctor - protected def newTaskImpl[R, Tp](b: Task[R, Tp]): TaskImpl[R, Tp] + protected def newWrappedTask[R, Tp](b: Task[R, Tp]): WrappedTask[R, Tp] /** The fork/join pool of this collection. */ def forkJoinPool: ForkJoinPool = environment.asInstanceOf[ForkJoinPool] - var environment: AnyRef = ForkJoinTasks.defaultForkJoinPool + val environment: ForkJoinPool /** Executes a task and does not wait for it to finish - instead returns a future. * * $fjdispatch */ def execute[R, Tp](task: Task[R, Tp]): () => R = { - val fjtask = newTaskImpl(task) + val fjtask = newWrappedTask(task) if (Thread.currentThread.isInstanceOf[ForkJoinWorkerThread]) { fjtask.fork @@ -483,7 +481,7 @@ trait ForkJoinTasks extends Tasks with HavingForkJoinPool { * @return the result of the task */ def executeAndWaitResult[R, Tp](task: Task[R, Tp]): R = { - val fjtask = newTaskImpl(task) + val fjtask = newWrappedTask(task) if (Thread.currentThread.isInstanceOf[ForkJoinWorkerThread]) { fjtask.fork @@ -513,28 +511,56 @@ object ForkJoinTasks { */ trait AdaptiveWorkStealingForkJoinTasks extends ForkJoinTasks with AdaptiveWorkStealingTasks { - class TaskImpl[R, Tp](val body: Task[R, Tp]) - extends super[ForkJoinTasks].TaskImpl[R, Tp] with super[AdaptiveWorkStealingTasks].TaskImpl[R, Tp] { - def split = body.split.map(b => newTaskImpl(b)) + class WrappedTask[R, Tp](val body: Task[R, Tp]) + extends super[ForkJoinTasks].WrappedTask[R, Tp] with super[AdaptiveWorkStealingTasks].WrappedTask[R, Tp] { + def split = body.split.map(b => newWrappedTask(b)) } - def newTaskImpl[R, Tp](b: Task[R, Tp]) = new TaskImpl[R, Tp](b) + def newWrappedTask[R, Tp](b: Task[R, Tp]) = new WrappedTask[R, Tp](b) } trait AdaptiveWorkStealingThreadPoolTasks extends ThreadPoolTasks with AdaptiveWorkStealingTasks { - class TaskImpl[R, Tp](val body: Task[R, Tp]) - extends super[ThreadPoolTasks].TaskImpl[R, Tp] with super[AdaptiveWorkStealingTasks].TaskImpl[R, Tp] { - def split = body.split.map(b => newTaskImpl(b)) + class WrappedTask[R, Tp](val body: Task[R, Tp]) + extends super[ThreadPoolTasks].WrappedTask[R, Tp] with super[AdaptiveWorkStealingTasks].WrappedTask[R, Tp] { + def split = body.split.map(b => newWrappedTask(b)) } - def newTaskImpl[R, Tp](b: Task[R, Tp]) = new TaskImpl[R, Tp](b) + def newWrappedTask[R, Tp](b: Task[R, Tp]) = new WrappedTask[R, Tp](b) } +trait ExecutionContextTasks extends Tasks { + + def executionContext = environment + + val environment: ExecutionContext + + // this part is a hack which allows switching + val driver: Tasks = executionContext match { + case eci: scala.concurrent.impl.ExecutionContextImpl => eci.executorService match { + case fjp: ForkJoinPool => new ForkJoinTaskSupport(fjp) + case tpe: ThreadPoolExecutor => new ThreadPoolTaskSupport(tpe) + case _ => ??? + } + case _ => ??? + } + + def execute[R, Tp](task: Task[R, Tp]): () => R = driver execute task + + def executeAndWaitResult[R, Tp](task: Task[R, Tp]): R = driver executeAndWaitResult task + + def parallelismLevel = driver.parallelismLevel + +} + + + + + diff --git a/src/library/scala/collection/parallel/immutable/ParHashMap.scala b/src/library/scala/collection/parallel/immutable/ParHashMap.scala index 7adf51cffb..266b179401 100644 --- a/src/library/scala/collection/parallel/immutable/ParHashMap.scala +++ b/src/library/scala/collection/parallel/immutable/ParHashMap.scala @@ -8,6 +8,8 @@ package scala.collection.parallel.immutable + + import scala.collection.parallel.ParMapLike import scala.collection.parallel.Combiner import scala.collection.parallel.IterableSplitter @@ -19,6 +21,9 @@ import scala.collection.generic.GenericParMapTemplate import scala.collection.generic.GenericParMapCompanion import scala.collection.immutable.{ HashMap, TrieIterator } import annotation.unchecked.uncheckedVariance +import collection.parallel.Task + + /** Immutable parallel hash map, based on hash tries. * @@ -153,7 +158,6 @@ private[parallel] abstract class HashMapCombiner[K, V] extends collection.parallel.BucketCombiner[(K, V), ParHashMap[K, V], (K, V), HashMapCombiner[K, V]](HashMapCombiner.rootsize) { //self: EnvironmentPassingCombiner[(K, V), ParHashMap[K, V]] => import HashMapCombiner._ - import collection.parallel.tasksupport._ val emptyTrie = HashMap.empty[K, V] def +=(elem: (K, V)) = { @@ -173,7 +177,7 @@ extends collection.parallel.BucketCombiner[(K, V), ParHashMap[K, V], (K, V), Has val bucks = buckets.filter(_ != null).map(_.headPtr) val root = new Array[HashMap[K, V]](bucks.length) - executeAndWaitResult(new CreateTrie(bucks, root, 0, bucks.length)) + combinerTaskSupport.executeAndWaitResult(new CreateTrie(bucks, root, 0, bucks.length)) var bitmap = 0 var i = 0 @@ -195,7 +199,7 @@ extends collection.parallel.BucketCombiner[(K, V), ParHashMap[K, V], (K, V), Has val bucks = buckets.filter(_ != null).map(_.headPtr) val root = new Array[HashMap[K, AnyRef]](bucks.length) - executeAndWaitResult(new CreateGroupedTrie(cbf, bucks, root, 0, bucks.length)) + combinerTaskSupport.executeAndWaitResult(new CreateGroupedTrie(cbf, bucks, root, 0, bucks.length)) var bitmap = 0 var i = 0 @@ -256,7 +260,7 @@ extends collection.parallel.BucketCombiner[(K, V), ParHashMap[K, V], (K, V), Has val fp = howmany / 2 List(new CreateTrie(bucks, root, offset, fp), new CreateTrie(bucks, root, offset + fp, howmany - fp)) } - def shouldSplitFurther = howmany > collection.parallel.thresholdFromSize(root.length, parallelismLevel) + def shouldSplitFurther = howmany > collection.parallel.thresholdFromSize(root.length, combinerTaskSupport.parallelismLevel) } class CreateGroupedTrie[Repr](cbf: () => Combiner[V, Repr], bucks: Array[Unrolled[(K, V)]], root: Array[HashMap[K, AnyRef]], offset: Int, howmany: Int) @@ -321,7 +325,7 @@ extends collection.parallel.BucketCombiner[(K, V), ParHashMap[K, V], (K, V), Has val fp = howmany / 2 List(new CreateGroupedTrie(cbf, bucks, root, offset, fp), new CreateGroupedTrie(cbf, bucks, root, offset + fp, howmany - fp)) } - def shouldSplitFurther = howmany > collection.parallel.thresholdFromSize(root.length, parallelismLevel) + def shouldSplitFurther = howmany > collection.parallel.thresholdFromSize(root.length, combinerTaskSupport.parallelismLevel) } } diff --git a/src/library/scala/collection/parallel/immutable/ParHashSet.scala b/src/library/scala/collection/parallel/immutable/ParHashSet.scala index 1cf0ccd391..0d7f04976e 100644 --- a/src/library/scala/collection/parallel/immutable/ParHashSet.scala +++ b/src/library/scala/collection/parallel/immutable/ParHashSet.scala @@ -8,6 +8,8 @@ package scala.collection.parallel.immutable + + import scala.collection.parallel.ParSetLike import scala.collection.parallel.Combiner import scala.collection.parallel.IterableSplitter @@ -19,6 +21,9 @@ import scala.collection.generic.GenericParTemplate import scala.collection.generic.GenericParCompanion import scala.collection.generic.GenericCompanion import scala.collection.immutable.{ HashSet, TrieIterator } +import collection.parallel.Task + + /** Immutable parallel hash set, based on hash tries. * @@ -127,7 +132,6 @@ private[immutable] abstract class HashSetCombiner[T] extends collection.parallel.BucketCombiner[T, ParHashSet[T], Any, HashSetCombiner[T]](HashSetCombiner.rootsize) { //self: EnvironmentPassingCombiner[T, ParHashSet[T]] => import HashSetCombiner._ - import collection.parallel.tasksupport._ val emptyTrie = HashSet.empty[T] def +=(elem: T) = { @@ -147,7 +151,7 @@ extends collection.parallel.BucketCombiner[T, ParHashSet[T], Any, HashSetCombine val bucks = buckets.filter(_ != null).map(_.headPtr) val root = new Array[HashSet[T]](bucks.length) - executeAndWaitResult(new CreateTrie(bucks, root, 0, bucks.length)) + combinerTaskSupport.executeAndWaitResult(new CreateTrie(bucks, root, 0, bucks.length)) var bitmap = 0 var i = 0 @@ -202,7 +206,7 @@ extends collection.parallel.BucketCombiner[T, ParHashSet[T], Any, HashSetCombine val fp = howmany / 2 List(new CreateTrie(bucks, root, offset, fp), new CreateTrie(bucks, root, offset + fp, howmany - fp)) } - def shouldSplitFurther = howmany > collection.parallel.thresholdFromSize(root.length, parallelismLevel) + def shouldSplitFurther = howmany > collection.parallel.thresholdFromSize(root.length, combinerTaskSupport.parallelismLevel) } } diff --git a/src/library/scala/collection/parallel/mutable/ParArray.scala b/src/library/scala/collection/parallel/mutable/ParArray.scala index 72a8184b10..5c3da66be0 100644 --- a/src/library/scala/collection/parallel/mutable/ParArray.scala +++ b/src/library/scala/collection/parallel/mutable/ParArray.scala @@ -21,6 +21,7 @@ import scala.collection.generic.Sizing import scala.collection.parallel.Combiner import scala.collection.parallel.SeqSplitter import scala.collection.parallel.ParSeqLike +import scala.collection.parallel.Task import scala.collection.parallel.CHECK_RATE import scala.collection.mutable.ArraySeq import scala.collection.mutable.Builder @@ -56,7 +57,6 @@ extends ParSeq[T] with Serializable { self => - import collection.parallel.tasksupport._ @transient private var array: Array[Any] = arrayseq.array.asInstanceOf[Array[Any]] @@ -584,22 +584,22 @@ self => val targetarr = targarrseq.array.asInstanceOf[Array[Any]] // fill it in parallel - executeAndWaitResult(new Map[S](f, targetarr, 0, length)) + tasksupport.executeAndWaitResult(new Map[S](f, targetarr, 0, length)) // wrap it into a parallel array (new ParArray[S](targarrseq)).asInstanceOf[That] } else super.map(f)(bf) override def scan[U >: T, That](z: U)(op: (U, U) => U)(implicit cbf: CanBuildFrom[ParArray[T], U, That]): That = - if (parallelismLevel > 1 && buildsArray(cbf(repr))) { + if (tasksupport.parallelismLevel > 1 && buildsArray(cbf(repr))) { // reserve an array val targarrseq = new ArraySeq[U](length + 1) val targetarr = targarrseq.array.asInstanceOf[Array[Any]] targetarr(0) = z // do a parallel prefix scan - if (length > 0) executeAndWaitResult(new CreateScanTree[U](0, size, z, op, splitter) mapResult { - tree => executeAndWaitResult(new ScanToArray(tree, z, op, targetarr)) + if (length > 0) tasksupport.executeAndWaitResult(new CreateScanTree[U](0, size, z, op, splitter) mapResult { + tree => tasksupport.executeAndWaitResult(new ScanToArray(tree, z, op, targetarr)) }) // wrap the array into a parallel array @@ -661,7 +661,7 @@ self => val fp = howmany / 2 List(new Map(f, targetarr, offset, fp), new Map(f, targetarr, offset + fp, howmany - fp)) } - def shouldSplitFurther = howmany > collection.parallel.thresholdFromSize(length, parallelismLevel) + def shouldSplitFurther = howmany > collection.parallel.thresholdFromSize(length, tasksupport.parallelismLevel) } /* serialization */ diff --git a/src/library/scala/collection/parallel/mutable/ParCtrie.scala b/src/library/scala/collection/parallel/mutable/ParCtrie.scala index 80add2407b..470972adad 100644 --- a/src/library/scala/collection/parallel/mutable/ParCtrie.scala +++ b/src/library/scala/collection/parallel/mutable/ParCtrie.scala @@ -13,6 +13,7 @@ package scala.collection.parallel.mutable import scala.collection.generic._ import scala.collection.parallel.Combiner import scala.collection.parallel.IterableSplitter +import scala.collection.parallel.Task import scala.collection.mutable.BasicNode import scala.collection.mutable.TNode import scala.collection.mutable.LNode @@ -40,8 +41,6 @@ extends ParMap[K, V] with ParCtrieCombiner[K, V] with Serializable { - import collection.parallel.tasksupport._ - def this() = this(new Ctrie) override def mapCompanion: GenericParMapCompanion[ParCtrie] = ParCtrie @@ -83,7 +82,7 @@ extends ParMap[K, V] case tn: TNode[_, _] => tn.cachedSize(ctrie) case ln: LNode[_, _] => ln.cachedSize(ctrie) case cn: CNode[_, _] => - executeAndWaitResult(new Size(0, cn.array.length, cn.array)) + tasksupport.executeAndWaitResult(new Size(0, cn.array.length, cn.array)) cn.cachedSize(ctrie) } } diff --git a/src/library/scala/collection/parallel/mutable/ParHashMap.scala b/src/library/scala/collection/parallel/mutable/ParHashMap.scala index 21a5b05749..6ce6c45460 100644 --- a/src/library/scala/collection/parallel/mutable/ParHashMap.scala +++ b/src/library/scala/collection/parallel/mutable/ParHashMap.scala @@ -17,6 +17,7 @@ import collection.mutable.DefaultEntry import collection.mutable.HashEntry import collection.mutable.HashTable import collection.mutable.UnrolledBuffer +import collection.parallel.Task @@ -156,8 +157,6 @@ private[mutable] abstract class ParHashMapCombiner[K, V](private val tableLoadFa extends collection.parallel.BucketCombiner[(K, V), ParHashMap[K, V], DefaultEntry[K, V], ParHashMapCombiner[K, V]](ParHashMapCombiner.numblocks) with collection.mutable.HashTable.HashUtils[K] { -//self: EnvironmentPassingCombiner[(K, V), ParHashMap[K, V]] => - import collection.parallel.tasksupport._ private var mask = ParHashMapCombiner.discriminantmask private var nonmasklen = ParHashMapCombiner.nonmasklength private var seedvalue = 27 @@ -179,7 +178,7 @@ extends collection.parallel.BucketCombiner[(K, V), ParHashMap[K, V], DefaultEntr // construct table val table = new AddingHashTable(size, tableLoadFactor, seedvalue) val bucks = buckets.map(b => if (b ne null) b.headPtr else null) - val insertcount = executeAndWaitResult(new FillBlocks(bucks, table, 0, bucks.length)) + val insertcount = combinerTaskSupport.executeAndWaitResult(new FillBlocks(bucks, table, 0, bucks.length)) table.setSize(insertcount) // TODO compare insertcount and size to see if compression is needed val c = table.hashTableContents @@ -300,7 +299,7 @@ extends collection.parallel.BucketCombiner[(K, V), ParHashMap[K, V], DefaultEntr override def merge(that: FillBlocks) { this.result += that.result } - def shouldSplitFurther = howmany > collection.parallel.thresholdFromSize(ParHashMapCombiner.numblocks, parallelismLevel) + def shouldSplitFurther = howmany > collection.parallel.thresholdFromSize(ParHashMapCombiner.numblocks, combinerTaskSupport.parallelismLevel) } } diff --git a/src/library/scala/collection/parallel/mutable/ParHashSet.scala b/src/library/scala/collection/parallel/mutable/ParHashSet.scala index 0217d0bfa8..e0a2ab03df 100644 --- a/src/library/scala/collection/parallel/mutable/ParHashSet.scala +++ b/src/library/scala/collection/parallel/mutable/ParHashSet.scala @@ -8,10 +8,15 @@ package scala.collection.parallel.mutable + + import collection.generic._ import collection.mutable.FlatHashTable import collection.parallel.Combiner import collection.mutable.UnrolledBuffer +import collection.parallel.Task + + /** A parallel hash set. * @@ -113,7 +118,6 @@ private[mutable] abstract class ParHashSetCombiner[T](private val tableLoadFacto extends collection.parallel.BucketCombiner[T, ParHashSet[T], Any, ParHashSetCombiner[T]](ParHashSetCombiner.numblocks) with collection.mutable.FlatHashTable.HashUtils[T] { //self: EnvironmentPassingCombiner[T, ParHashSet[T]] => - import collection.parallel.tasksupport._ private var mask = ParHashSetCombiner.discriminantmask private var nonmasklen = ParHashSetCombiner.nonmasklength private var seedvalue = 27 @@ -139,7 +143,7 @@ with collection.mutable.FlatHashTable.HashUtils[T] { private def parPopulate: FlatHashTable.Contents[T] = { // construct it in parallel val table = new AddingFlatHashTable(size, tableLoadFactor, seedvalue) - val (inserted, leftovers) = executeAndWaitResult(new FillBlocks(buckets, table, 0, buckets.length)) + val (inserted, leftovers) = combinerTaskSupport.executeAndWaitResult(new FillBlocks(buckets, table, 0, buckets.length)) var leftinserts = 0 for (elem <- leftovers) leftinserts += table.insertEntry(0, table.tableLength, elem.asInstanceOf[T]) table.setSize(leftinserts + inserted) @@ -304,7 +308,7 @@ with collection.mutable.FlatHashTable.HashUtils[T] { // the total number of successfully inserted elements is adjusted accordingly result = (this.result._1 + that.result._1 + inserted, remainingLeftovers concat that.result._2) } - def shouldSplitFurther = howmany > collection.parallel.thresholdFromSize(ParHashMapCombiner.numblocks, parallelismLevel) + def shouldSplitFurther = howmany > collection.parallel.thresholdFromSize(ParHashMapCombiner.numblocks, combinerTaskSupport.parallelismLevel) } } diff --git a/src/library/scala/collection/parallel/mutable/ResizableParArrayCombiner.scala b/src/library/scala/collection/parallel/mutable/ResizableParArrayCombiner.scala index eadc93d422..01eb17024e 100644 --- a/src/library/scala/collection/parallel/mutable/ResizableParArrayCombiner.scala +++ b/src/library/scala/collection/parallel/mutable/ResizableParArrayCombiner.scala @@ -8,18 +8,20 @@ package scala.collection.parallel.mutable + + import scala.collection.generic.Sizing import scala.collection.mutable.ArraySeq import scala.collection.mutable.ArrayBuffer import scala.collection.parallel.TaskSupport -//import scala.collection.parallel.EnvironmentPassingCombiner import scala.collection.parallel.unsupportedop import scala.collection.parallel.Combiner +import scala.collection.parallel.Task + + /** An array combiner that uses a chain of arraybuffers to store elements. */ trait ResizableParArrayCombiner[T] extends LazyCombiner[T, ParArray[T], ExposedArrayBuffer[T]] { -//self: EnvironmentPassingCombiner[T, ParArray[T]] => - import collection.parallel.tasksupport._ override def sizeHint(sz: Int) = if (chain.length == 1) chain(0).sizeHint(sz) @@ -30,7 +32,7 @@ trait ResizableParArrayCombiner[T] extends LazyCombiner[T, ParArray[T], ExposedA val arrayseq = new ArraySeq[T](size) val array = arrayseq.array.asInstanceOf[Array[Any]] - executeAndWaitResult(new CopyChainToArray(array, 0, size)) + combinerTaskSupport.executeAndWaitResult(new CopyChainToArray(array, 0, size)) new ParArray(arrayseq) } else { // optimisation if there is only 1 array @@ -79,7 +81,7 @@ trait ResizableParArrayCombiner[T] extends LazyCombiner[T, ParArray[T], ExposedA val fp = howmany / 2 List(new CopyChainToArray(array, offset, fp), new CopyChainToArray(array, offset + fp, howmany - fp)) } - def shouldSplitFurther = howmany > collection.parallel.thresholdFromSize(size, parallelismLevel) + def shouldSplitFurther = howmany > collection.parallel.thresholdFromSize(size, combinerTaskSupport.parallelismLevel) } } diff --git a/src/library/scala/collection/parallel/mutable/UnrolledParArrayCombiner.scala b/src/library/scala/collection/parallel/mutable/UnrolledParArrayCombiner.scala index dc583fb4e7..410b542a68 100644 --- a/src/library/scala/collection/parallel/mutable/UnrolledParArrayCombiner.scala +++ b/src/library/scala/collection/parallel/mutable/UnrolledParArrayCombiner.scala @@ -18,9 +18,9 @@ import scala.collection.mutable.ArrayBuffer import scala.collection.mutable.UnrolledBuffer import scala.collection.mutable.UnrolledBuffer.Unrolled import scala.collection.parallel.TaskSupport -//import scala.collection.parallel.EnvironmentPassingCombiner import scala.collection.parallel.unsupportedop import scala.collection.parallel.Combiner +import scala.collection.parallel.Task @@ -40,8 +40,6 @@ extends Combiner[T, ParArray[T]] { // because size is doubling, random access is O(logn)! val buff = new DoublingUnrolledBuffer[Any] - import collection.parallel.tasksupport._ - def +=(elem: T) = { buff += elem this @@ -51,7 +49,7 @@ extends Combiner[T, ParArray[T]] { val arrayseq = new ArraySeq[T](size) val array = arrayseq.array.asInstanceOf[Array[Any]] - executeAndWaitResult(new CopyUnrolledToArray(array, 0, size)) + combinerTaskSupport.executeAndWaitResult(new CopyUnrolledToArray(array, 0, size)) new ParArray(arrayseq) } @@ -109,7 +107,7 @@ extends Combiner[T, ParArray[T]] { val fp = howmany / 2 List(new CopyUnrolledToArray(array, offset, fp), new CopyUnrolledToArray(array, offset + fp, howmany - fp)) } - def shouldSplitFurther = howmany > collection.parallel.thresholdFromSize(size, parallelismLevel) + def shouldSplitFurther = howmany > collection.parallel.thresholdFromSize(size, combinerTaskSupport.parallelismLevel) override def toString = "CopyUnrolledToArray(" + offset + ", " + howmany + ")" } } diff --git a/src/library/scala/collection/parallel/package.scala b/src/library/scala/collection/parallel/package.scala index f5e05ffefb..943e0208c7 100644 --- a/src/library/scala/collection/parallel/package.scala +++ b/src/library/scala/collection/parallel/package.scala @@ -46,8 +46,16 @@ package object parallel { else new ThreadPoolTaskSupport } else new ThreadPoolTaskSupport - val tasksupport = getTaskSupport - + val defaultTaskSupport: TaskSupport = getTaskSupport + + def setTaskSupport[Coll](c: Coll, t: TaskSupport): Coll = { + c match { + case pc: ParIterableLike[_, _, _] => pc.tasksupport = t + case _ => // do nothing + } + c + } + /* implicit conversions */ implicit def factory2ops[From, Elem, To](bf: CanBuildFrom[From, Elem, To]) = new FactoryOps[From, Elem, To] { diff --git a/src/library/scala/concurrent/ConcurrentPackageObject.scala b/src/library/scala/concurrent/ConcurrentPackageObject.scala index ae17c7e032..7d005838d3 100644 --- a/src/library/scala/concurrent/ConcurrentPackageObject.scala +++ b/src/library/scala/concurrent/ConcurrentPackageObject.scala @@ -8,22 +8,30 @@ package scala.concurrent + + +import java.util.concurrent.{ Executors, ExecutorService } +import scala.concurrent.forkjoin.ForkJoinPool import scala.util.{ Duration, Try, Success, Failure } import ConcurrentPackageObject._ + + /** This package object contains primitives for concurrent and parallel programming. */ abstract class ConcurrentPackageObject { /** A global execution environment for executing lightweight tasks. */ lazy val executionContext = - new impl.ExecutionContextImpl(java.util.concurrent.Executors.newCachedThreadPool()) - - /** A global service for scheduling tasks for execution. - */ - // lazy val scheduler = - // new default.SchedulerImpl - + new impl.ExecutionContextImpl(getExecutorService) + + private[concurrent] def getExecutorService: AnyRef = + if (util.Properties.isJavaAtLeast("1.6")) { + val vendor = util.Properties.javaVmVendor + if ((vendor contains "Oracle") || (vendor contains "Sun") || (vendor contains "Apple")) new ForkJoinPool + else Executors.newCachedThreadPool() + } else Executors.newCachedThreadPool() + val handledFutureException: PartialFunction[Throwable, Throwable] = { case t: Throwable if isFutureThrowable(t) => t } diff --git a/src/library/scala/concurrent/impl/ExecutionContextImpl.scala b/src/library/scala/concurrent/impl/ExecutionContextImpl.scala index 7b44d02612..7984aa02b7 100644 --- a/src/library/scala/concurrent/impl/ExecutionContextImpl.scala +++ b/src/library/scala/concurrent/impl/ExecutionContextImpl.scala @@ -11,20 +11,25 @@ package scala.concurrent.impl import java.util.concurrent.{Callable, ExecutorService} +import scala.concurrent.forkjoin._ import scala.concurrent.{ExecutionContext, resolver, Awaitable, body2awaitable} import scala.util.{ Duration, Try, Success, Failure } import scala.collection.mutable.Stack -class ExecutionContextImpl(executorService: ExecutorService) extends ExecutionContext { +class ExecutionContextImpl(val executorService: AnyRef) extends ExecutionContext { import ExecutionContextImpl._ def execute(runnable: Runnable): Unit = executorService match { - // case fj: ForkJoinPool => - // TODO fork if more applicable - // executorService execute runnable - case _ => + case fj: ForkJoinPool => + if (Thread.currentThread.isInstanceOf[ForkJoinWorkerThread]) { + val fjtask = ForkJoinTask.adapt(runnable) + fjtask.fork + } else { + fj.execute(runnable) + } + case executorService: ExecutorService => executorService execute runnable } |