You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by li...@apache.org on 2018/08/31 17:43:34 UTC
spark git commit: [SPARK-25286][CORE] Removing the dangerous parmap
Repository: spark
Updated Branches:
refs/heads/master 7fc8881b0 -> 32da87dfa
[SPARK-25286][CORE] Removing the dangerous parmap
## What changes were proposed in this pull request?
I propose to remove one of `parmap` methods which accepts an execution context as a parameter. The method should be removed to eliminate any deadlocks that can occur if `parmap` is called recursively on thread pools restricted by size.
Closes #22292 from MaxGekk/remove-overloaded-parmap.
Authored-by: Maxim Gekk <ma...@databricks.com>
Signed-off-by: Xiao Li <ga...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/32da87df
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/32da87df
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/32da87df
Branch: refs/heads/master
Commit: 32da87dfa451fff677ed9316f740be2abdbff6a4
Parents: 7fc8881
Author: Maxim Gekk <ma...@databricks.com>
Authored: Fri Aug 31 10:43:30 2018 -0700
Committer: Xiao Li <ga...@gmail.com>
Committed: Fri Aug 31 10:43:30 2018 -0700
----------------------------------------------------------------------
.../scala/org/apache/spark/rdd/UnionRDD.scala | 17 ++++++-----
.../org/apache/spark/util/ThreadUtils.scala | 32 +++-----------------
.../streaming/util/FileBasedWriteAheadLog.scala | 5 +--
3 files changed, 16 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/32da87df/core/src/main/scala/org/apache/spark/rdd/UnionRDD.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/rdd/UnionRDD.scala b/core/src/main/scala/org/apache/spark/rdd/UnionRDD.scala
index 4b6f732..60e383a 100644
--- a/core/src/main/scala/org/apache/spark/rdd/UnionRDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/UnionRDD.scala
@@ -20,13 +20,12 @@ package org.apache.spark.rdd
import java.io.{IOException, ObjectOutputStream}
import scala.collection.mutable.ArrayBuffer
-import scala.concurrent.ExecutionContext
+import scala.collection.parallel.ForkJoinTaskSupport
import scala.concurrent.forkjoin.ForkJoinPool
import scala.reflect.ClassTag
import org.apache.spark.{Dependency, Partition, RangeDependency, SparkContext, TaskContext}
import org.apache.spark.annotation.DeveloperApi
-import org.apache.spark.util.ThreadUtils.parmap
import org.apache.spark.util.Utils
/**
@@ -60,7 +59,8 @@ private[spark] class UnionPartition[T: ClassTag](
}
object UnionRDD {
- private[spark] lazy val threadPool = new ForkJoinPool(8)
+ private[spark] lazy val partitionEvalTaskSupport =
+ new ForkJoinTaskSupport(new ForkJoinPool(8))
}
@DeveloperApi
@@ -74,13 +74,14 @@ class UnionRDD[T: ClassTag](
rdds.length > conf.getInt("spark.rdd.parallelListingThreshold", 10)
override def getPartitions: Array[Partition] = {
- val partitionLengths = if (isPartitionListingParallel) {
- implicit val ec = ExecutionContext.fromExecutor(UnionRDD.threadPool)
- parmap(rdds)(_.partitions.length)
+ val parRDDs = if (isPartitionListingParallel) {
+ val parArray = rdds.par
+ parArray.tasksupport = UnionRDD.partitionEvalTaskSupport
+ parArray
} else {
- rdds.map(_.partitions.length)
+ rdds
}
- val array = new Array[Partition](partitionLengths.sum)
+ val array = new Array[Partition](parRDDs.map(_.partitions.length).seq.sum)
var pos = 0
for ((rdd, rddIndex) <- rdds.zipWithIndex; split <- rdd.partitions) {
array(pos) = new UnionPartition(pos, rdd, rddIndex, split.index)
http://git-wip-us.apache.org/repos/asf/spark/blob/32da87df/core/src/main/scala/org/apache/spark/util/ThreadUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/ThreadUtils.scala b/core/src/main/scala/org/apache/spark/util/ThreadUtils.scala
index f0e5add..cb0c205 100644
--- a/core/src/main/scala/org/apache/spark/util/ThreadUtils.scala
+++ b/core/src/main/scala/org/apache/spark/util/ThreadUtils.scala
@@ -284,36 +284,12 @@ private[spark] object ThreadUtils {
try {
implicit val ec = ExecutionContext.fromExecutor(pool)
- parmap(in)(f)
+ val futures = in.map(x => Future(f(x)))
+ val futureSeq = Future.sequence(futures)
+
+ awaitResult(futureSeq, Duration.Inf)
} finally {
pool.shutdownNow()
}
}
-
- /**
- * Transforms input collection by applying the given function to each element in parallel fashion.
- * Comparing to the map() method of Scala parallel collections, this method can be interrupted
- * at any time. This is useful on canceling of task execution, for example.
- *
- * @param in - the input collection which should be transformed in parallel.
- * @param f - the lambda function will be applied to each element of `in`.
- * @param ec - an execution context for parallel applying of the given function `f`.
- * @tparam I - the type of elements in the input collection.
- * @tparam O - the type of elements in resulted collection.
- * @return new collection in which each element was given from the input collection `in` by
- * applying the lambda function `f`.
- */
- def parmap[I, O, Col[X] <: TraversableLike[X, Col[X]]]
- (in: Col[I])
- (f: I => O)
- (implicit
- cbf: CanBuildFrom[Col[I], Future[O], Col[Future[O]]], // For in.map
- cbf2: CanBuildFrom[Col[Future[O]], O, Col[O]], // for Future.sequence
- ec: ExecutionContext
- ): Col[O] = {
- val futures = in.map(x => Future(f(x)))
- val futureSeq = Future.sequence(futures)
-
- awaitResult(futureSeq, Duration.Inf)
- }
}
http://git-wip-us.apache.org/repos/asf/spark/blob/32da87df/streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala
----------------------------------------------------------------------
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala b/streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala
index bba071e..f0161e1 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala
@@ -312,10 +312,11 @@ private[streaming] object FileBasedWriteAheadLog {
handler: I => Iterator[O]): Iterator[O] = {
val taskSupport = new ExecutionContextTaskSupport(executionContext)
val groupSize = taskSupport.parallelismLevel.max(8)
- implicit val ec = executionContext
source.grouped(groupSize).flatMap { group =>
- ThreadUtils.parmap(group)(handler)
+ val parallelCollection = group.par
+ parallelCollection.tasksupport = taskSupport
+ parallelCollection.map(handler)
}.flatten
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org