diff options
author | Holden Karau <holden@us.ibm.com> | 2016-03-04 10:56:58 +0000 |
---|---|---|
committer | Sean Owen <sowen@cloudera.com> | 2016-03-04 10:56:58 +0000 |
commit | c04dc27cedd3d75781fda4c24da16b6ada44d3e4 (patch) | |
tree | 70f0f5fa93c9dae9f13a31e24b54a18fee0e5977 /streaming/src/main | |
parent | 27e88faa058c1364d0e99fffc0c5cb64ef817bd3 (diff) | |
download | spark-c04dc27cedd3d75781fda4c24da16b6ada44d3e4.tar.gz spark-c04dc27cedd3d75781fda4c24da16b6ada44d3e4.tar.bz2 spark-c04dc27cedd3d75781fda4c24da16b6ada44d3e4.zip |
[SPARK-13398][STREAMING] Move away from thread pool task support to forkjoin
## What changes were proposed in this pull request?
Remove old deprecated ThreadPoolExecutor and replace with ExecutionContext using a ForkJoinPool. The downside of this is that scala's ForkJoinPool doesn't give us a way to specify the thread pool name (and is a wrapper of Java's in 2.12) except by providing a custom factory. Note that we can't use Java's ForkJoinPool directly in Scala 2.11 since it uses a ExecutionContext which reports system parallelism. One other implicit change that happens is the old ExecutionContext would have reported a different default parallelism since it used system parallelism rather than threadpool parallelism (this was likely not intended but also likely not a huge difference).
The previous version of this PR attempted to use an execution context constructed on the ThreadPool (but not the deprecated ThreadPoolExecutor class) so as to keep the ability to have human readable named threads but this reported system parallelism.
## How was this patch tested?
unit tests: streaming/testOnly org.apache.spark.streaming.util.*
Author: Holden Karau <holden@us.ibm.com>
Closes #11423 from holdenk/SPARK-13398-move-away-from-ThreadPoolTaskSupport-java-forkjoin.
Diffstat (limited to 'streaming/src/main')
-rw-r--r-- | streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala | 23 |
1 files changed, 12 insertions, 11 deletions
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 314263f26e..a3b7e783ac 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 @@ -18,11 +18,11 @@ package org.apache.spark.streaming.util import java.nio.ByteBuffer import java.util.{Iterator => JIterator} -import java.util.concurrent.{RejectedExecutionException, ThreadPoolExecutor} +import java.util.concurrent.RejectedExecutionException import scala.collection.JavaConverters._ import scala.collection.mutable.ArrayBuffer -import scala.collection.parallel.ThreadPoolTaskSupport +import scala.collection.parallel.ExecutionContextTaskSupport import scala.concurrent.{Await, ExecutionContext, Future} import scala.language.postfixOps @@ -62,8 +62,8 @@ private[streaming] class FileBasedWriteAheadLog( private val threadpoolName = { "WriteAheadLogManager" + callerName.map(c => s" for $c").getOrElse("") } - private val threadpool = ThreadUtils.newDaemonCachedThreadPool(threadpoolName, 20) - private val executionContext = ExecutionContext.fromExecutorService(threadpool) + private val forkJoinPool = ThreadUtils.newForkJoinPool(threadpoolName, 20) + private val executionContext = ExecutionContext.fromExecutorService(forkJoinPool) override protected def logName = { getClass.getName.stripSuffix("$") + @@ -144,7 +144,7 @@ private[streaming] class FileBasedWriteAheadLog( } else { // For performance gains, it makes sense to parallelize the recovery if // closeFileAfterWrite = true - seqToParIterator(threadpool, logFilesToRead, readFile).asJava + seqToParIterator(executionContext, logFilesToRead, readFile).asJava } } @@ -283,16 +283,17 @@ private[streaming] object FileBasedWriteAheadLog { /** * This creates an iterator from a parallel collection, by keeping at most `n` objects in memory - * at any given time, where `n` is the size of the thread pool. This is crucial for use cases - * where we create `FileBasedWriteAheadLogReader`s during parallel recovery. We don't want to - * open up `k` streams altogether where `k` is the size of the Seq that we want to parallelize. + * at any given time, where `n` is at most the max of the size of the thread pool or 8. This is + * crucial for use cases where we create `FileBasedWriteAheadLogReader`s during parallel recovery. + * We don't want to open up `k` streams altogether where `k` is the size of the Seq that we want + * to parallelize. */ def seqToParIterator[I, O]( - tpool: ThreadPoolExecutor, + executionContext: ExecutionContext, source: Seq[I], handler: I => Iterator[O]): Iterator[O] = { - val taskSupport = new ThreadPoolTaskSupport(tpool) - val groupSize = tpool.getMaximumPoolSize.max(8) + val taskSupport = new ExecutionContextTaskSupport(executionContext) + val groupSize = taskSupport.parallelismLevel.max(8) source.grouped(groupSize).flatMap { group => val parallelCollection = group.par parallelCollection.tasksupport = taskSupport |