diff options
-rw-r--r-- | src/actors/scala/actors/scheduler/ForkJoinScheduler.scala | 30 |
1 files changed, 8 insertions, 22 deletions
diff --git a/src/actors/scala/actors/scheduler/ForkJoinScheduler.scala b/src/actors/scala/actors/scheduler/ForkJoinScheduler.scala index bbcbfa90f4..86fb14db50 100644 --- a/src/actors/scala/actors/scheduler/ForkJoinScheduler.scala +++ b/src/actors/scala/actors/scheduler/ForkJoinScheduler.scala @@ -1,7 +1,6 @@ package scala.actors package scheduler -import java.lang.Thread.State import java.util.{Collection, ArrayList} import scala.concurrent.forkjoin._ @@ -12,9 +11,9 @@ import scala.concurrent.forkjoin._ */ class ForkJoinScheduler(val initCoreSize: Int, val maxSize: Int, daemon: Boolean) extends Runnable with IScheduler with TerminationMonitor { - private var pool = makeNewPool() - private var terminating = false - private var snapshoting = false + private var pool = makeNewPool() // guarded by this + private var terminating = false // guarded by this + private var snapshoting = false // guarded by this // this has to be a java.util.Collection, since this is what // the ForkJoinPool returns. @@ -55,15 +54,6 @@ class ForkJoinScheduler(val initCoreSize: Int, val maxSize: Int, daemon: Boolean } } - private def allWorkersBlocked: Boolean = - (pool.workers != null) && - pool.workers.forall(t => { - (t == null) || { - val s = t.getState() - s == State.BLOCKED || s == State.WAITING || s == State.TIMED_WAITING - } - }) - override def run() { try { while (true) { @@ -78,18 +68,12 @@ class ForkJoinScheduler(val initCoreSize: Int, val maxSize: Int, daemon: Boolean throw new QuitException if (allTerminated) { - //Debug.info(this+": all actors terminated") + Debug.info(this+": all actors terminated") throw new QuitException } if (!snapshoting) { gc() - - // check if we need more threads to avoid deadlock - val poolSize = pool.getPoolSize() - if (allWorkersBlocked && (poolSize < ThreadPoolConfig.maxPoolSize)) { - pool.setParallelism(poolSize + 1) - } } else if (pool.isQuiescent()) { val list = new ArrayList[ForkJoinTask[_]] val num = pool.drainTasksTo(list) @@ -142,8 +126,9 @@ class ForkJoinScheduler(val initCoreSize: Int, val maxSize: Int, daemon: Boolean terminating = true } - def isActive = + def isActive = synchronized { (pool ne null) && !pool.isShutdown() + } override def managedBlock(blocker: scala.concurrent.ManagedBlocker) { ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker { @@ -170,8 +155,9 @@ class ForkJoinScheduler(val initCoreSize: Int, val maxSize: Int, daemon: Boolean error("scheduler is still active") else snapshoting = false + + pool = makeNewPool() } - pool = makeNewPool() val iter = drainedTasks.iterator() while (iter.hasNext()) { pool.execute(iter.next()) |