regression: parallel collections hang #8955
Comments
|
Imported From: https://issues.scala-lang.org/browse/SI-8955?orig=1 |
|
@retronym said: /cc @viktorklang |
|
@retronym said: % git diff -U50
diff --git a/src/library/scala/concurrent/impl/ExecutionContextImpl.scala b/src/library/scala/concurrent/impl/ExecutionContextImpl.scala
index 32f30b9..627161a 100644
--- a/src/library/scala/concurrent/impl/ExecutionContextImpl.scala
+++ b/src/library/scala/concurrent/impl/ExecutionContextImpl.scala
@@ -1,94 +1,94 @@
/* __ *\
** ________ ___ / / ___ Scala API **
** / __/ __// _ | / / / _ | (c) 2003-2013, LAMP/EPFL **
** __\ \/ /__/ __ |/ /__/ __ | http://scala-lang.org/ **
** /____/\___/_/ |_/____/_/ | | **
** |/ **
\* */
package scala.concurrent.impl
import java.util.concurrent.{ LinkedBlockingQueue, Callable, Executor, ExecutorService, Executors, ThreadFactory, TimeUnit, ThreadPoolExecutor }
import java.util.concurrent.atomic.AtomicInteger
import java.util.Collection
import scala.concurrent.forkjoin._
import scala.concurrent.{ BlockContext, ExecutionContext, Awaitable, CanAwait, ExecutionContextExecutor, ExecutionContextExecutorService }
import scala.util.control.NonFatal
import scala.annotation.tailrec
private[scala] class ExecutionContextImpl private[impl] (val executor: Executor, val reporter: Throwable => Unit) extends ExecutionContextExecutor {
require(executor ne null, "Executor must not be null")
override def execute(runnable: Runnable) = executor execute runnable
override def reportFailure(t: Throwable) = reporter(t)
}
private[concurrent] object ExecutionContextImpl {
// Implement BlockContext on FJP threads
final class DefaultThreadFactory(
daemonic: Boolean,
maxThreads: Int,
prefix: String,
uncaught: Thread.UncaughtExceptionHandler) extends ThreadFactory with ForkJoinPool.ForkJoinWorkerThreadFactory {
require(prefix ne null, "DefaultThreadFactory.prefix must be non null")
require(maxThreads > 0, "DefaultThreadFactory.maxThreads must be greater than 0")
private final val currentNumberOfThreads = new AtomicInteger(0)
@tailrec private final def reserveThread(): Boolean = currentNumberOfThreads.get() match {
- case `maxThreads` | Int.`MaxValue` => false
+ case `maxThreads` | Int.`MaxValue` => true
case other => currentNumberOfThreads.compareAndSet(other, other + 1) || reserveThread()
}
@tailrec private final def deregisterThread(): Boolean = currentNumberOfThreads.get() match {
case 0 => false
case other => currentNumberOfThreads.compareAndSet(other, other - 1) || deregisterThread()
}
def wire[T <: Thread](thread: T): T = {
thread.setDaemon(daemonic)
thread.setUncaughtExceptionHandler(uncaught)
thread.setName(prefix + "-" + thread.getId())
thread
}
// As per ThreadFactory contract newThread should return `null` if cannot create new thread.
def newThread(runnable: Runnable): Thread =
if (reserveThread())
wire(new Thread(new Runnable {
// We have to decrement the current thread count when the thread exits
override def run() = try runnable.run() finally deregisterThread()
})) else null
def newThread(fjp: ForkJoinPool): ForkJoinWorkerThread =
if (reserveThread()) {
wire(new ForkJoinWorkerThread(fjp) with BlockContext {
// We have to decrement the current thread count when the thread exits
final override def onTermination(exception: Throwable): Unit = deregisterThread()
final override def blockOn[T](thunk: =>T)(implicit permission: CanAwait): T = {
var result: T = null.asInstanceOf[T]
ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker {
@volatile var isdone = false
override def block(): Boolean = {
result = try {
// When we block, switch out the BlockContext temporarily so that nested blocking does not created N new Threads
BlockContext.withBlockContext(BlockContext.defaultBlockContext) { thunk }
} finally {
isdone = true
}
true
}
override def isReleasable = isdone
})
result
}
})
} else null
}
This might indicate an upstream bug in ForkJoin triggered by a thread factory that occasionally returns |
|
Viktor Klang (viktorklang) said: If you pass in the following System property: -Dscala.concurrent.context.maxThreads=x100 Does it still hang? If not, then I have an idea for an improvement (the max number of blocked threads is set too conservative with the updated SIP14) |
|
@retronym said: Without that, here's the stack trace. |
|
Viktor Klang (viktorklang) said: diff --git a/src/library/scala/concurrent/impl/ExecutionContextImpl.scala b/src/library/scala/concurrent/impl/ExecutionContextImpl.scala
The number 256 is going to be the default for the max threads for FJP in Java9 (down from 32k) so this change will harmonize the settings while making it possible to override from the outside. The cause of the deadlock is twofold: 1) The test uses ExecutionContext.global, which is not designed for typical ForkJoin workloads since it has async = true (FIFO instead of LIFO) 2) And we capped the default max number of threads to be created when doing managed blocking from 32k to number of cores (a tad too aggressive it seems) |
|
@retronym said: |
|
Viktor Klang (viktorklang) said: given that there are only a couple of people on the planet who understand the ForkJoinX code, and the only relevant change that would impact ParCol (at least that I made) was to cap the max threads of the pool (uncapped it's 32k), we have to either conclude that: A) The failing test is somehow broken. B) It needs to be able to spawn more threads than it currently is allowed to C) ForkJoin is somehow broken For A) I think Aleks is probably the best debugger |
|
Viktor Klang (viktorklang) said: |
|
Viktor Klang (viktorklang) said: I'm not ruling out a potential bug in: https://github.com/scala/scala/blob/2.11.x/src/library/scala/collection/parallel/Tasks.scala#L465 |
|
@retronym said: import scala.annotation.unchecked.uncheckedVariance
import scala.concurrent.forkjoin._
object Test {
val size = 10000
val parallelismLevel = 8
class Foreach[S](val sz: Int) extends Task {
def leaf() = ()
def shouldSplitFurther = sz > 1
def split = if (sz < 2) Seq(this) else Seq(new Foreach[S](sz / 2), new Foreach[S](sz / 2))
}
val pool = {
// val defaultPool = new ForkJoinPool() // okay
val maxThreads = parallelismLevel // parallelismLevel * 2 appears to be okay
val limitedThreadPool = new ForkJoinPool(parallelismLevel, new DefaultThreadFactory(maxThreads), Thread.getDefaultUncaughtExceptionHandler, true)
limitedThreadPool
}
def test() {
val fjtask = new WrappedTask(new Foreach(size))
pool.execute(fjtask)
fjtask.sync()
print(".")
}
def main(args: Array[String]): Unit = {
val repeats = 100000
for (i <- 1 to repeats) test()
println("Done.")
sys.exit()
}
}
trait Task {
def leaf()
def shouldSplitFurther: Boolean
def split: Seq[Task]
def tryLeaf() {
leaf()
}
}
class WrappedTask(val body: Task) extends RecursiveTask[Unit] {
@volatile var next: WrappedTask = null
@volatile var shouldWaitFor = true
def start() = fork
def sync() = join
def tryCancel() = tryUnfork
def release() {}
def split: Seq[WrappedTask] = body.split.map(b => new WrappedTask(b))
def compute() = if (body.shouldSplitFurther) {
internal()
release()
} else {
body.tryLeaf()
release()
}
def internal() = {
var last = spawnSubtasks()
last.body.tryLeaf()
last.release()
while (last.next != null) {
last = last.next
if (last.tryCancel()) {
last.body.tryLeaf()
last.release()
} else {
last.sync()
}
}
}
def spawnSubtasks() = {
var last: WrappedTask = null
var head: WrappedTask = this
do {
val subtasks = head.split
head = subtasks.head
for (t <- subtasks.tail.reverse) {
t.next = last
last = t
t.start()
}
} while (head.body.shouldSplitFurther)
head.next = last
head
}
}
class DefaultThreadFactory(maxThreads: Int) extends ForkJoinPool.ForkJoinWorkerThreadFactory {
val numThreads = new java.util.concurrent.atomic.AtomicInteger(0)
def reserveThread(): Boolean = {
val n = numThreads.incrementAndGet()
n <= maxThreads
}
def newThread(runnable: Runnable): Thread =
if (reserveThread()) new Thread(runnable) else null
override def newThread(fjp: ForkJoinPool): ForkJoinWorkerThread =
if (reserveThread()) new ForkJoinWorkerThread(fjp) {} else null
} |
|
Viktor Klang (viktorklang) said: |
|
Viktor Klang (viktorklang) said: import scala.annotation.unchecked.uncheckedVariance
import scala.concurrent.forkjoin._
trait Task {
def leaf()
def shouldSplitFurther: Boolean
def split: Seq[Task]
def tryLeaf() {
leaf()
}
}
class WrappedTask(val body: Task) extends RecursiveTask[Unit] {
@volatile var next: WrappedTask = null
@volatile var shouldWaitFor = true
def start() = fork
def sync() = join
def tryCancel() = tryUnfork
def release() {}
def split: Seq[WrappedTask] = body.split.map(b => new WrappedTask(b))
def compute() = {
if (body.shouldSplitFurther) internal()
else body.tryLeaf()
release()
}
def internal() = {
var last = spawnSubtasks()
last.body.tryLeaf()
last.release()
while (last.next != null) {
last = last.next
if (last.tryCancel()) {
last.body.tryLeaf()
last.release()
} else {
last.sync()
}
}
}
def spawnSubtasks() = {
var last: WrappedTask = null
var head: WrappedTask = this
do {
val subtasks = head.split
head = subtasks.head
for (t <- subtasks.tail.reverse) {
t.next = last
last = t
t.start()
}
} while (head.body.shouldSplitFurther)
head.next = last
head
}
}
class DefaultThreadFactory(val maxThreads: Int) extends ForkJoinPool.ForkJoinWorkerThreadFactory {
val currentNumberOfThreads = new java.util.concurrent.atomic.AtomicInteger(0)
val hasMaxedOut = new java.util.concurrent.atomic.AtomicInteger(0)
private final def deregisterThread(): Boolean = currentNumberOfThreads.get() match {
case 0 => false
case other => currentNumberOfThreads.compareAndSet(other, other - 1) || deregisterThread()
}
private final def reserveThread(): Boolean = currentNumberOfThreads.get() match {
case `maxThreads` | Int.`MaxValue` => false
case other => currentNumberOfThreads.compareAndSet(other, other + 1) || reserveThread()
}
override def newThread(fjp: ForkJoinPool): ForkJoinWorkerThread =
if (reserveThread()) new ForkJoinWorkerThread(fjp) {
final override def onTermination(exception: Throwable): Unit =
if (!deregisterThread()) new Exception("Couldn't deregister worker!!!").printStackTrace(System.err)
else println("Releasing worker: " + this)
} else {
if(hasMaxedOut.compareAndSet(0, 1))
new Exception("Tried to spawn new worker but couldn't").printStackTrace(System.err)
null
}
}
object Test {
val size = 10000
val parallelismLevel = 8
class Foreach[S](val sz: Int) extends Task {
def leaf() = ()
def shouldSplitFurther = sz > 1
def split = if (sz < 2) Seq(this) else Seq(new Foreach[S](sz / 2), new Foreach[S](sz / 2))
}
val pool = new ForkJoinPool(parallelismLevel, new DefaultThreadFactory(parallelismLevel / 2), null, false)
def test() {
val fjtask = new WrappedTask(new Foreach(size))
pool.execute(fjtask)
fjtask.sync()
print('.')
}
def main(args: Array[String]): Unit = {
val repeats = 100000
for (i <- 1 to repeats) test()
println("Done.")
sys.exit()
}
}
Test.main(null) |
|
@retronym said: |
|
@retronym said: |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
This hangs reliably for me:
I'm bisecting to find the point of regression.
The text was updated successfully, but these errors were encountered: